"""Workspace-scoped, immutable methods and append-only validation evidence.

The ledger consumes existing task proofs. It never starts browser/model work.
Reliability is recalculated from effective events, never incremented counters.
Legacy learned_entries remain an explicitly unverified historical projection.
"""
from __future__ import annotations

from collections import Counter
from copy import deepcopy
from datetime import datetime, timezone
import hashlib
import json
import re
from typing import Any

from psycopg.types.json import Jsonb

from backend.db import database as db
from backend.prospecting.contracts import BUSINESS_ID, KNOWLEDGE_ID

RULE_VERSION = "independent-evidence-v1"
OUTCOMES = {"success", "contradiction", "blocked", "program_error", "unused", "unknown"}
_ALIASES = {"confirmed": "success", "contradicted": "contradiction", "failure": "contradiction",
            "platform_block": "blocked", "error": "program_error", "insufficient": "unknown"}
_DIMENSIONS = ("platform", "stage", "capability", "source_kind", "page_variant")
_SPEC_KEYS = {"key", "version", "title", "kind", "capability", "source_kind", "platform", "stage",
              "page_variant", "preconditions", "parameters", "steps", "postcondition", "recovery",
              "dependencies", "origin", "branches", "terminal", "required_actions", "cost_hint"}
_PRIVATE_KEYS = {"quote", "nickname", "author", "author_name", "profile_url", "source_url", "url",
                 "account", "account_id", "user_id", "message_text", "user_notes", "keywords"}
_EVENT_METADATA = {"verified", "verifier", "reason_code", "run_id", "action_id", "goal_version",
                   "executed_branch", "terminal_complete", "dependency_event_ids", "model_tokens",
                   "cost", "operation_count", "supersedes", "program_verdict"}


def _json(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str)


def _digest(value: Any, prefix: str = "") -> str:
    return prefix + hashlib.sha256(_json(value).encode()).hexdigest()[:40]


def _projection(value: Any) -> Any:
    """Keep public ledger results safe for task JSONB, prompts and API clients."""
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, dict):
        return {key: _projection(child) for key, child in value.items()}
    if isinstance(value, list):
        return [_projection(child) for child in value]
    return value


def _clean_identifier(value: Any, field: str, limit: int = 200) -> str:
    text = str(value or "").strip()
    if not text or len(text) > limit or not re.fullmatch(r"[\w.:/\-]+", text):
        raise ValueError(f"invalid knowledge {field}")
    return text


def _check_generic(value: Any) -> None:
    """Method bodies contain parameterized instructions, never task source data."""
    if isinstance(value, dict):
        if _PRIVATE_KEYS.intersection(value):
            raise ValueError("personal/source fields belong in task evidence, not shared methods")
        for child in value.values():
            _check_generic(child)
    elif isinstance(value, list):
        for child in value:
            _check_generic(child)
    elif isinstance(value, str):
        if len(value) > 5000 or re.search(r"https?://", value):
            raise ValueError("shared methods must use parameters and task proof references")


def normalize_spec(raw: dict) -> dict:
    if not isinstance(raw, dict):
        raise ValueError("method spec must be an object")
    unknown = set(raw) - _SPEC_KEYS - {"id", "expected_result", "deps", "source_task_id"}
    if unknown:
        raise ValueError(f"unknown method fields: {sorted(unknown)}")
    spec = {key: deepcopy(value) for key, value in raw.items() if key in _SPEC_KEYS}
    spec["key"] = _clean_identifier(raw.get("key"), "key")
    spec["version"] = _clean_identifier(raw.get("version", "1"), "version", 40)
    for key in _DIMENSIONS:
        spec[key] = _clean_identifier(raw.get(key, "any"), key, 128)
    spec["kind"] = str(raw.get("kind", "operation"))
    if spec["kind"] not in {"operation", "pipeline", "strategy"}:
        raise ValueError("method kind must be operation, pipeline or strategy")
    spec["origin"] = str(raw.get("origin", "model"))
    if spec["origin"] not in {"program_template", "observed", "model", "preset", "historical"}:
        raise ValueError("invalid method origin")
    spec["title"] = str(raw.get("title") or spec["key"])[:240]
    for key in ("preconditions", "parameters", "steps", "recovery", "branches", "required_actions"):
        spec.setdefault(key, [])
    spec["postcondition"] = deepcopy(raw.get("postcondition", raw.get("expected_result", {})))
    spec["dependencies"] = deepcopy(raw.get("dependencies", raw.get("deps", [])))
    if not isinstance(spec["dependencies"], list):
        raise ValueError("dependencies must list exact method ids or key/version pairs")
    for dep in spec["dependencies"]:
        if isinstance(dep, str):
            _clean_identifier(dep, "dependency")
        elif not isinstance(dep, dict) or not (dep.get("id") or dep.get("key") and dep.get("version")):
            raise ValueError("dependency must identify its exact version")
    if spec["kind"] == "pipeline" and not spec.get("terminal"):
        raise ValueError("pipeline requires an explicit terminal postcondition")
    _check_generic(spec)
    return spec


def _method_row(row: dict) -> dict:
    spec = deepcopy(row.get("spec") or row)
    spec["id"] = row.get("id") or _digest([spec["key"], spec["version"]], "method_")
    for field in ("created_at", "created_from_task_id"):
        if row.get(field) is not None:
            spec[field] = row[field]
    return _projection(spec)


def register_method(spec: dict, task_id: str = "") -> dict:
    """Register immutable content; a content change requires a new version."""
    clean = normalize_spec(spec)
    method_id = _digest([clean["key"], clean["version"]], "method_")
    with db.get_conn() as conn:
        db._execute(conn, """INSERT INTO prospecting_methods
            (workspace_id,id,method_key,version,spec,created_from_task_id)
            VALUES (%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING""",
            (db._sid(), method_id, clean["key"], clean["version"], Jsonb(clean), str(task_id or "")))
        row = db._fetchone(conn, "SELECT * FROM prospecting_methods WHERE workspace_id=%s AND id=%s",
                           (db._sid(), method_id))
        if not row or row["spec"] != clean:
            raise ValueError("method content changed: register a new version")
    return _method_row(row)


def _refs(raw: dict | None) -> dict:
    raw = raw or {}
    if not isinstance(raw, dict):
        raise ValueError("evidence_refs must be an object of task proof references")
    refs = {}
    for key, aliases in {"pre": ("pre", "before", "precondition"),
                         "post": ("post", "after", "postcondition")}.items():
        value = next((raw[a] for a in aliases if raw.get(a)), None)
        if value is not None:
            values = value if isinstance(value, list) else [value]
            if not all(isinstance(item, str) and 0 < len(item) <= 512
                       and re.fullmatch(r"[\w.:/\-]+", item) and not re.search(r"https?://", item)
                       for item in values):
                raise ValueError("evidence must reference task artifacts, not copy source text/URLs")
            refs[key] = deepcopy(value)
    return refs


def prepare_event(method: dict, task_id: str, object_key: str, outcome: str,
                  evidence_refs: dict | None = None, context: dict | None = None,
                  provenance: str = "program", event_key: str | None = None,
                  supersedes: str | None = None, **metadata: Any) -> dict:
    """Pure validation boundary used by record_event and fixture tests."""
    requested = _ALIASES.get(str(outcome), str(outcome))
    if requested not in OUTCOMES:
        raise ValueError("invalid knowledge validation outcome")
    _clean_identifier(task_id, "task_id", 128)
    if not object_key:
        raise ValueError("validation requires a stable task object")
    refs = _refs(evidence_refs)
    supplied = context or {}
    scene = {field: str(supplied.get(field) or method.get(field) or "any") for field in _DIMENSIONS}
    meta = {key: deepcopy(value) for key, value in metadata.items() if key in _EVENT_METADATA}
    for key in _EVENT_METADATA:
        if key in supplied and key not in meta:
            meta[key] = deepcopy(supplied[key])
    verdict = str(meta.get("program_verdict") or "")
    verified = meta.get("verified") is True or verdict == requested
    effective, reason = requested, str(meta.get("reason_code") or "")
    if requested in {"success", "contradiction"}:
        if verdict in {"success", "contradiction"} and verdict != requested:
            effective, reason = "unknown", "program_verdict_mismatch"
        elif provenance != "program" or not verified or not meta.get("verifier") or not refs.get("pre") or not refs.get("post"):
            effective, reason = "unknown", "program_proof_required"
        elif method.get("kind") == "strategy":
            effective, reason = "unknown", "business_strategy_is_not_technical_validation"
        elif method.get("kind") == "pipeline" and requested == "success":
            branch = meta.get("executed_branch")
            allowed = {str(item.get("id")) if isinstance(item, dict) else str(item)
                       for item in method.get("branches", [])}
            if not meta.get("terminal_complete") or not branch or allowed and branch not in allowed:
                effective, reason = "unknown", "pipeline_terminal_and_actual_branch_required"
            elif method.get("dependencies") and not meta.get("dependency_event_ids"):
                effective, reason = "unknown", "pipeline_dependency_proofs_required"
    meta["reason_code"] = reason
    payload = {"method_id": method["id"], "method_version": str(method.get("version", "1")),
               "task_id": task_id, "object_digest": _digest(str(object_key), "object_"),
               "requested_outcome": requested, "outcome": effective, "evidence_refs": refs,
               "context": scene, "provenance": str(provenance), "metadata": meta,
               "supersedes": str(supersedes or ""), "rule_version": RULE_VERSION}
    payload["event_key"] = str(event_key or _digest(payload, "validation_"))
    if len(payload["event_key"]) > 200:
        raise ValueError("event_key too long")
    payload["id"] = _digest([method["id"], payload["event_key"]], "validation_")
    return payload


def _event_row(row: dict) -> dict:
    result = deepcopy(row.get("payload") or row)
    if row.get("created_at") is not None:
        result["created_at"] = row["created_at"]
    return _projection(result)


def record_event(method_id: str, task_id: str, object_key: str, outcome: str,
                 evidence_refs: dict | None = None, context: dict | None = None,
                 provenance: str = "program", event_key: str | None = None,
                 supersedes: str | None = None, **metadata: Any) -> dict:
    with db.get_conn() as conn:
        row = db._fetchone(conn, "SELECT * FROM prospecting_methods WHERE workspace_id=%s AND id=%s",
                           (db._sid(), str(method_id)))
        if not row:
            raise ValueError("method does not exist in this workspace")
        event = prepare_event(_method_row(row), task_id, object_key, outcome, evidence_refs, context,
                              provenance, event_key, supersedes, **metadata)
        # Compare the original normalized call, before dependency projection.
        # Later corrections can invalidate dependencies without rewriting the
        # first immutable receipt or turning a harmless replay into an error.
        input_digest = _digest(event, "input_")
        existing = db._fetchone(conn, "SELECT * FROM prospecting_validation_events WHERE workspace_id=%s AND id=%s",
                                (db._sid(), event["id"]))
        if existing:
            if existing["payload"].get("input_digest") == input_digest or existing["payload"] == event:
                return _event_row(existing)
            raise ValueError("validation key already has different evidence; append a superseding correction")
        event["input_digest"] = input_digest
        if supersedes:
            old = db._fetchone(conn, "SELECT * FROM prospecting_validation_events WHERE workspace_id=%s AND id=%s",
                               (db._sid(), str(supersedes)))
            if (not old or old["method_id"] != method_id or old["task_id"] != task_id
                    or old["object_digest"] != event["object_digest"]):
                raise ValueError("correction must supersede this method/task/object's existing event")
        _verify_dependency_events(conn, _method_row(row), event)
        db._execute(conn, """INSERT INTO prospecting_validation_events
            (workspace_id,id,method_id,task_id,object_digest,event_key,payload)
            VALUES (%s,%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING""",
            (db._sid(), event["id"], method_id, task_id, event["object_digest"], event["event_key"], Jsonb(event)))
        saved = db._fetchone(conn, "SELECT * FROM prospecting_validation_events WHERE workspace_id=%s AND id=%s",
                             (db._sid(), event["id"]))
        if not saved or saved["payload"].get("input_digest") != input_digest:
            raise ValueError("validation key already has different evidence; append a superseding correction")
    return _event_row(saved)


def _verify_dependency_events(conn, method: dict, event: dict) -> None:
    if method.get("kind") != "pipeline" or event["outcome"] != "success" or not method.get("dependencies"):
        return
    rows = db._fetchall(conn, """SELECT * FROM prospecting_validation_events
        WHERE workspace_id=%s AND task_id=%s""", (db._sid(), event["task_id"]))
    if not _pipeline_proofs_valid(method, event, [_event_row(row) for row in rows]):
        event["outcome"] = "unknown"
        event["metadata"]["reason_code"] = "pipeline_dependency_proofs_invalid"


def _pipeline_proofs_valid(method: dict, event: dict, events: list[dict]) -> bool:
    """An old component proof corrected later also invalidates its chain proof."""
    if method.get("kind") != "pipeline":
        return True
    meta = event.get("metadata") or {}
    branch = meta.get("executed_branch")
    allowed = {str(item.get("id")) if isinstance(item, dict) else str(item) for item in method.get("branches", [])}
    if not meta.get("terminal_complete") or not branch or allowed and branch not in allowed:
        return False
    needed = {_dependency_id(dep) for dep in method.get("dependencies", [])}
    if not needed:
        return True
    refs = meta.get("dependency_event_ids") or []
    if not isinstance(refs, list):
        return False
    active = effective_events(events)
    proven = {item["method_id"] for item in active
              if item["id"] in refs and item["task_id"] == event["task_id"]
              and item.get("outcome") == "success" and item.get("provenance") == "program"
              and item.get("metadata", {}).get("verifier") and item.get("metadata", {}).get("verified") is True
              and item.get("evidence_refs", {}).get("pre") and item.get("evidence_refs", {}).get("post")}
    return needed.issubset(proven)


def effective_events(events: list[dict]) -> list[dict]:
    """Keep all evidence but exclude superseded facts from the current projection."""
    unique = {str(event["id"]): event for event in events}
    superseded = {str(event["supersedes"]) for event in unique.values() if event.get("supersedes")}
    return sorted((event for key, event in unique.items() if key not in superseded), key=_event_order)


def _event_order(event: dict) -> tuple[str, str]:
    return (str(event.get("created_at") or event.get("at") or ""), str(event.get("id") or ""))


def _dependency_id(dep: Any) -> str:
    if isinstance(dep, str):
        return dep
    return str(dep.get("id") or _digest([str(dep["key"]), str(dep["version"])], "method_"))


def applicability(method: dict, context: dict | None = None) -> tuple[bool, list[str]]:
    context = context or {}
    reasons = []
    for field in _DIMENSIONS:
        actual, expected = context.get(field), method.get(field, "any")
        if actual and expected not in {"any", "*", actual}:
            reasons.append(f"{field}_mismatch")
    wanted_version = context.get("method_version")
    if wanted_version is not None and str(method.get("version")) != str(wanted_version):
        reasons.append("method_version_mismatch")
    available = context.get("dependency_versions", context.get("available_method_ids"))
    if method.get("dependencies"):
        if available is None:
            reasons.append("dependency_versions_unconfirmed")
        else:
            if isinstance(available, dict):
                available_ids = set(available) | {_digest([str(k), str(v)], "method_") for k, v in available.items()}
            else:
                available_ids = set(available)
            if any(_dependency_id(dep) not in available_ids for dep in method["dependencies"]):
                reasons.append("dependency_version_mismatch")
    required = set(method.get("required_actions") or [])
    if method.get("capability") in {"follow", "message", "send_message"}:
        required.add("message" if method["capability"] == "send_message" else method["capability"])
    if required and not required.issubset(set(context.get("authorized_actions") or [])):
        reasons.append("action_not_authorized")
    return not reasons, reasons or ["scope_matches"]


def _same_scene(event: dict, method: dict, context: dict) -> bool:
    for field in _DIMENSIONS:
        expected = context.get(field) or method.get(field)
        observed = (event.get("context") or {}).get(field) or method.get(field)
        if expected and expected not in {"any", "*"} and observed != expected:
            return False
    return str(event.get("method_version", method.get("version"))) == str(method.get("version"))


def _independent_sample(events: list[dict]) -> list[dict]:
    """Deterministic task/object matching, reserving contradictions before successes.

    Each task and stable object contribute at most once. Contradiction priority
    prevents a favorable retry being selected instead of an unresolved failure.
    Within each outcome maximum bipartite matching avoids greedy undercounting.
    """
    chosen, used_tasks, used_objects = [], set(), set()
    for outcome in ("contradiction", "success"):
        edges = {}
        for event in sorted(events, key=_event_order):
            task, obj = event["task_id"], event["object_digest"]
            if event["outcome"] == outcome and task not in used_tasks and obj not in used_objects:
                edges.setdefault(task, {}).setdefault(obj, event)
        owners = {}

        def assign(task, visited):
            for obj in sorted(edges[task]):
                if obj in visited:
                    continue
                visited.add(obj)
                if obj not in owners or assign(owners[obj], visited):
                    owners[obj] = task
                    return True
            return False

        for task in sorted(edges):
            assign(task, set())
        for obj, task in sorted(owners.items()):
            chosen.append(edges[task][obj])
            used_tasks.add(task)
            used_objects.add(obj)
    return sorted(chosen, key=_event_order)


def evaluate_method(method: dict, events: list[dict], context: dict | None = None) -> dict:
    """Pure, explainable view. Internal score is intentionally absent from UI."""
    context = context or {}
    method = _method_row(method)
    active = effective_events([event for event in events if event.get("method_id") == method["id"]])
    scene = [event for event in active if _same_scene(event, method, context)]
    # prepared program events are the sole source of technical facts; raw/model
    # reports cannot bypass the validation boundary by entering a restored list.
    eligible = [event for event in scene if event.get("outcome") in {"success", "contradiction"}
                and event.get("provenance") == "program" and event.get("metadata", {}).get("verifier")
                and (event.get("metadata", {}).get("verified") is True
                     or str(event.get("metadata", {}).get("program_verdict")) in {"success", "contradiction"})
                and event.get("evidence_refs", {}).get("pre") and event.get("evidence_refs", {}).get("post")
                and (event["outcome"] == "contradiction" or _pipeline_proofs_valid(method, event, events))]
    sample = _independent_sample(eligible)
    counts = Counter(event["outcome"] for event in sample)
    auxiliary = Counter(event.get("outcome", "unknown") for event in scene)
    unaccepted = sum(event.get("outcome") in {"success", "contradiction"} and event not in eligible for event in scene)
    stats = {"success": counts["success"], "contradiction": counts["contradiction"],
             **{outcome: auxiliary[outcome] for outcome in ("blocked", "program_error", "unused", "unknown")},
             "independent_tasks": len({event["task_id"] for event in sample}),
             "independent_objects": len({event["object_digest"] for event in sample}),
             "coverage_tasks": len({event["task_id"] for event in eligible}),
             "coverage_objects": len({event["object_digest"] for event in eligible})}
    stats["unknown"] += unaccepted
    verified_last = max(sample, key=_event_order) if sample else None
    # Only independent technical samples influence recency/cost tie breaks.
    # Repeated unknowns or receipts must not improve a method's ranking.
    hints = method.get("cost_hint") if isinstance(method.get("cost_hint"), dict) else {}
    observed_cost = {}
    for field in ("model_tokens", "operation_count"):
        values = [event.get("metadata", {}).get(field) for event in sample]
        values = [float(value) for value in values if isinstance(value, (int, float)) and value >= 0]
        fallback = hints.get(field)
        observed_cost[field] = sum(values) / len(values) if values else (fallback if isinstance(fallback, (int, float)) else None)
    applicable, reasons = applicability(method, context)
    contradiction = any(event["outcome"] == "contradiction" for event in eligible)
    if contradiction:
        status = "paused"
        reasons = [*reasons, "unresolved_contradiction"]
    elif not applicable:
        status = "paused"
    elif counts["success"] >= 3:
        status = "preferred"
    elif counts["success"]:
        status = "trial"
    else:
        status = "exploration"
    last = max(scene, key=_event_order) if scene else None
    result = {**method, "status": status, "applicability": applicable and not contradiction,
              "match_reasons": reasons, "stats": stats, "rule_version": RULE_VERSION,
              "sample_event_ids": [event["id"] for event in sample],
              "ranking_evidence": {"last_verified_at": str(verified_last.get("created_at") or verified_last.get("at") or "") if verified_last else "",
                                   "complete_samples": len(sample), "observed_cost": observed_cost},
              "historical_unverified": False, "provenance": method.get("origin", "model"),
              "expected_result": method.get("postcondition", {}),
              "last_validation": {"at": str(last.get("created_at") or last.get("at") or ""),
                                  "outcome": last["outcome"], "reason": last.get("metadata", {}).get("reason_code", ""),
                                  "verifier": last.get("metadata", {}).get("verifier", ""),
                                  "event_id": last["id"], "task_id": last["task_id"]} if last else None,
              "evidence_refs": [{"task_id": event["task_id"], "event_id": event["id"],
                                  **event.get("evidence_refs", {}),
                                  **({"action_id": event["metadata"]["action_id"]}
                                     if event.get("metadata", {}).get("action_id") else {})}
                                 for event in scene[-20:] if event.get("evidence_refs")]}
    return result


def _ranking_key(method: dict) -> tuple:
    tiers = {"preferred": 0, "trial": 1, "exploration": 2, "paused": 3, "legacy": 4}
    stats = method["stats"]
    score = (stats["success"] + 1) / (stats["success"] + stats["contradiction"] + 2)
    evidence = method.get("ranking_evidence") or {}
    # ISO timestamps preserve recency ordering. A numeric value is easier to
    # reverse without exposing the uncalibrated ranking score to the UI.
    try:
        observed = datetime.fromisoformat(evidence.get("last_verified_at", "").replace("Z", "+00:00"))
        recent = (observed if observed.tzinfo else observed.replace(tzinfo=timezone.utc)).timestamp()
    except (ValueError, TypeError):
        recent = 0
    completeness = evidence.get("complete_samples", 0)
    cost = evidence.get("observed_cost") or {}
    def known_cost(field):
        value = cost.get(field)
        return float(value) if isinstance(value, (int, float)) else float("inf")
    return (not method.get("applicability"), tiers[method["status"]], -score, -completeness,
            -recent, -stats.get("coverage_objects", 0), known_cost("model_tokens"),
            known_cost("operation_count"), method["key"], str(method["version"]))


def rank_methods(methods: list[dict], events: list[dict], context: dict | None = None) -> list[dict]:
    return sorted((evaluate_method(method, events, context) for method in methods), key=_ranking_key)


def _read_ledger(conn) -> tuple[list[dict], list[dict]]:
    methods = db._fetchall(conn, "SELECT * FROM prospecting_methods WHERE workspace_id=%s ORDER BY id", (db._sid(),))
    events = db._fetchall(conn, "SELECT * FROM prospecting_validation_events WHERE workspace_id=%s ORDER BY created_at,id", (db._sid(),))
    return [_method_row(row) for row in methods], [_event_row(row) for row in events]


def _legacy_rows(conn) -> list[dict]:
    rows = db._fetchall(conn, """SELECT id,key,category,source_task_id,updated_at FROM learned_entries
        WHERE school_id=%s AND knowledge_id=%s AND business_id=%s ORDER BY id""",
        (db._sid(), KNOWLEDGE_ID, BUSINESS_ID))
    return [{"id": f"legacy_{row['id']}", "key": row["key"], "title": row["key"], "version": "historical",
             "kind": "operation", "status": "legacy", "historical_unverified": True,
             "provenance": "legacy", "capability": "unknown", "source_kind": "unknown",
             "applicability": False, "match_reasons": ["historical_validation_incomplete"],
             "stats": {field: None for field in ("success", "contradiction", "blocked", "program_error", "unused",
                        "unknown", "independent_tasks", "independent_objects", "coverage_tasks", "coverage_objects")},
             "last_validation": None, "evidence_refs": [], "steps": [], "expected_result": {}, "recovery": []}
            for row in rows]


def list_methods(context: dict | None = None, include_legacy: bool = True) -> list[dict]:
    with db.get_conn() as conn:
        methods, events = _read_ledger(conn)
        keys = {method.get("key") for method in methods}
        legacy = [row for row in _legacy_rows(conn) if row.get("key") not in keys] if include_legacy else []
    return rank_methods(methods, events, context) + legacy


def select_methods(context: dict, limit: int = 8) -> list[dict]:
    """Only applicable current methods. No task is created to raise a score."""
    return [method for method in list_methods(context, include_legacy=False)
            if method["applicability"] and method["status"] in {"preferred", "trial", "exploration"}
            and method.get("kind") != "strategy"][:max(0, int(limit))]


def record_business_feedback(task_id: str, result_version: str, feedback: dict, *, goal_version: Any = None,
                             event_key: str | None = None, source: str = "user") -> dict:
    """Versioned business observations never mutate technical reliability."""
    _clean_identifier(task_id, "task_id", 128)
    if not isinstance(feedback, dict):
        raise ValueError("business feedback must be an object")
    payload = {"task_id": task_id, "result_version": str(result_version), "goal_version": goal_version,
               "feedback": deepcopy(feedback), "source": str(source), "attribution": "unproven"}
    identifier = _digest([task_id, event_key or payload], "feedback_")
    with db.get_conn() as conn:
        db._execute(conn, """INSERT INTO prospecting_business_feedback
            (workspace_id,id,task_id,result_version,payload) VALUES (%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING""",
            (db._sid(), identifier, task_id, str(result_version), Jsonb(payload)))
        row = db._fetchone(conn, "SELECT * FROM prospecting_business_feedback WHERE workspace_id=%s AND id=%s",
                           (db._sid(), identifier))
        if not row or row["payload"] != payload:
            raise ValueError("feedback key already records another result; append a new event")
    return _projection({"id": row["id"], **row["payload"], "created_at": row["created_at"]})


def task_summary(task_id: str) -> dict:
    with db.get_conn() as conn:
        methods, all_events = _read_ledger(conn)
        feedback = db._fetchall(conn, """SELECT * FROM prospecting_business_feedback
            WHERE workspace_id=%s AND task_id=%s ORDER BY created_at,id""", (db._sid(), task_id))
    task_events = [event for event in all_events if event["task_id"] == task_id]
    identifiers = {event["method_id"] for event in task_events}
    return {"task_id": task_id, "rule_version": RULE_VERSION,
            "methods": rank_methods([method for method in methods if method["id"] in identifiers], all_events),
            "events": task_events, "effective_event_ids": [event["id"] for event in effective_events(task_events)],
            "outcomes": dict(Counter(event["outcome"] for event in effective_events(task_events))),
            "business_feedback": [_projection({"id": row["id"], **row["payload"], "created_at": row["created_at"]}) for row in feedback]}