aios/active-runtime/app/task_plane/lineage_store.py

"""Actual TaskPlane binding for task identity, lineage and root review.

All writes use the existing :class:`TaskPlaneStore` connection and transaction
fence.  Request metadata is never used as an authorization source.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable, Mapping, Sequence

if __package__ and "." in __package__:  # app.task_plane package import
    from ..aios_task_lineage import (
        AcceptanceCondition,
        AcceptanceEvaluator,
        NextLegalAction,
        RelationType,
        RootAcceptanceSnapshot,
        RootReviewRecord,
        TaskRecord as LineageTaskRecord,
        lineage_from_metadata,
        normalize_goal_slug,
        root_review_from_record,
    )
    from ..aios_task_lineage_integration import (
        DISPLAY_NAME_METADATA_KEY,
        LINEAGE_METADATA_KEY,
        LINEAGE_REQUEST_DIGEST_KEY,
        NewTaskIdentityRequest,
        PERSISTED_DISPATCH_STATUSES,
        RootActionDispatchReceipt,
        TaskAuthoritySnapshot,
        TaskIdentityLineageService,
        TerminalReviewOutcome,
        _as_task_record,
    )
    from ..aios_task_lineage_taskplane_adapter import (
        AuthenticatedTaskContext,
        InProcessTaskPlaneLineageAdapter,
        TaskPlaneServiceBindings,
    )
    from ..aios_task_lineage_hooks import TaskLineageLifecycleHooks
else:  # top-level task_plane import used by direct-module test harnesses
    from aios_task_lineage import (
        AcceptanceCondition,
        AcceptanceEvaluator,
        NextLegalAction,
        RelationType,
        RootAcceptanceSnapshot,
        RootReviewRecord,
        TaskRecord as LineageTaskRecord,
        lineage_from_metadata,
        normalize_goal_slug,
        root_review_from_record,
    )
    from aios_task_lineage_integration import (
        DISPLAY_NAME_METADATA_KEY,
        LINEAGE_METADATA_KEY,
        LINEAGE_REQUEST_DIGEST_KEY,
        NewTaskIdentityRequest,
        PERSISTED_DISPATCH_STATUSES,
        RootActionDispatchReceipt,
        TaskAuthoritySnapshot,
        TaskIdentityLineageService,
        TerminalReviewOutcome,
        _as_task_record,
    )
    from aios_task_lineage_taskplane_adapter import (
        AuthenticatedTaskContext,
        InProcessTaskPlaneLineageAdapter,
        TaskPlaneServiceBindings,
    )
    from aios_task_lineage_hooks import TaskLineageLifecycleHooks

from .delegation import DelegationEngine
from .errors import ConflictError, ForbiddenError, NotFoundError, ValidationError
from .models import TaskPlan, TaskRecord, canonical_json, compute_sha256_hex
from .store import TaskPlaneStore


RESERVED_LINEAGE_METADATA_KEYS = frozenset(
    {LINEAGE_METADATA_KEY, DISPLAY_NAME_METADATA_KEY, LINEAGE_REQUEST_DIGEST_KEY}
)
PROVEN_TERMINAL_STATES = frozenset({"COMPLETED", "FAILED_FINAL", "CANCELLED"})


@dataclass(frozen=True)
class PrincipalLineageBinding:
    """Server-owned projection of one authenticated service identity."""

    authenticated_identity: str
    owner_principal_id: str
    mission_id: str
    preferred_owner_alias: str | None
    root_acceptance_ref: str

    def __post_init__(self) -> None:
        for name in (
            "authenticated_identity",
            "owner_principal_id",
            "mission_id",
            "root_acceptance_ref",
        ):
            if not str(getattr(self, name)).strip():
                raise ValueError(f"{name} is required")


class TaskLineagePolicy:
    """Immutable service policy for authenticated identity and acceptance."""

    def __init__(
        self,
        *,
        principal_bindings: Sequence[PrincipalLineageBinding],
        acceptance_evaluators: Mapping[str, AcceptanceEvaluator],
        acceptance_evaluator_resolver: Callable[
            [str], AcceptanceEvaluator | None
        ]
        | None = None,
    ) -> None:
        self._bindings: dict[str, PrincipalLineageBinding] = {}
        for binding in principal_bindings:
            if binding.authenticated_identity in self._bindings:
                raise ValueError("duplicate authenticated lineage identity")
            self._bindings[binding.authenticated_identity] = binding
        self._evaluators = dict(acceptance_evaluators)
        self._acceptance_evaluator_resolver = acceptance_evaluator_resolver

    def with_acceptance_evaluator_resolver(
        self, resolver: Callable[[str], AcceptanceEvaluator | None]
    ) -> "TaskLineagePolicy":
        previous = self._acceptance_evaluator_resolver

        def combined(reference: str) -> AcceptanceEvaluator | None:
            resolved = resolver(reference)
            if resolved is not None:
                return resolved
            return previous(reference) if previous is not None else None

        return TaskLineagePolicy(
            principal_bindings=tuple(self._bindings.values()),
            acceptance_evaluators=self._evaluators,
            acceptance_evaluator_resolver=combined,
        )

    def resolve_principal(self, authenticated_identity: str) -> PrincipalLineageBinding:
        try:
            return self._bindings[authenticated_identity]
        except KeyError as exc:
            raise ForbiddenError("authenticated identity has no lineage policy binding") from exc

    def _acceptance_evaluator(self, reference: str) -> AcceptanceEvaluator | None:
        evaluator = self._evaluators.get(reference)
        if evaluator is None and self._acceptance_evaluator_resolver is not None:
            evaluator = self._acceptance_evaluator_resolver(reference)
        return evaluator

    def assert_acceptance_ref_registered(self, reference: str) -> None:
        if self._acceptance_evaluator(reference) is None:
            raise ForbiddenError("root acceptance policy is not registered")

    def evaluate_acceptance(
        self,
        root: LineageTaskRecord,
        descendants: Sequence[LineageTaskRecord],
    ) -> RootAcceptanceSnapshot:
        reference = root.lineage.root_acceptance_ref
        evaluator = self._acceptance_evaluator(reference)
        if evaluator is None:
            raise ForbiddenError("root acceptance policy is not registered")
        snapshot = evaluator(root, descendants)
        if not isinstance(snapshot, RootAcceptanceSnapshot) or not snapshot.conditions:
            return RootAcceptanceSnapshot(
                (
                    AcceptanceCondition(
                        key="root-acceptance-proof-unavailable",
                        satisfied=False,
                        owner_boundary=True,
                    ),
                )
            )
        normalized: list[AcceptanceCondition] = []
        for condition in snapshot.conditions:
            if condition.satisfied and not condition.evidence_refs:
                normalized.append(
                    AcceptanceCondition(
                        key=condition.key,
                        satisfied=False,
                        owner_boundary=True,
                    )
                )
            else:
                normalized.append(condition)
        return RootAcceptanceSnapshot(tuple(normalized))


class TaskPlaneLineageStoreBinding:
    """Official in-process read/write surface over the existing TaskPlane DB."""

    def __init__(self, store: TaskPlaneStore) -> None:
        self.store = store

    def service_bindings(self) -> TaskPlaneServiceBindings:
        return TaskPlaneServiceBindings(
            get_task_snapshot=self.get_task_snapshot,
            list_descendant_snapshots=self.list_descendant_snapshots,
            reserve_sibling_sequence=self.reserve_sibling_sequence,
            compare_and_swap_task_metadata=self.compare_and_swap_task_metadata,
            get_root_review_record=self.get_root_review_record,
            compare_and_swap_root_review_record=self.compare_and_swap_root_review_record,
            get_root_action_dispatch=self.get_root_action_dispatch,
            dispatch_root_action=self.dispatch_root_action,
        )

    def _terminal_proof(self, task: TaskRecord) -> Mapping[str, object] | None:
        row = self.store.conn.execute(
            "SELECT * FROM task_terminal_proofs WHERE task_id=?", (task.task_id,)
        ).fetchone()
        if row is None or task.status not in PROVEN_TERMINAL_STATES:
            return None
        claim = self.store.conn.execute(
            "SELECT status,result_payload FROM claims WHERE claim_id=? AND task_id=?",
            (row["claim_id"], task.task_id),
        ).fetchone()
        if claim is None or claim["status"] != "COMPLETED" or not claim["result_payload"]:
            return None
        try:
            result = json.loads(claim["result_payload"])
            refs = tuple(json.loads(row["evidence_refs_json"]))
        except (TypeError, ValueError, json.JSONDecodeError):
            return None
        if result.get("result_digest") != row["result_digest"]:
            return None
        if task.revision < int(row["task_revision"]):
            return None
        evidence_ids = {entry["evidence_id"] for entry in self.store.list_evidence(task.task_id)}
        if not set(refs).issubset(evidence_ids):
            return None
        if not self.store.verify_event_chain(task.task_id):
            return None
        terminal_event = any(
            event["type"] in {"TASK_COMPLETED", "TASK_FAILED"}
            and event["data"].get("claim_id") == row["claim_id"]
            for event in self.store.list_events(task.task_id)
        )
        if not terminal_event or not self.store.verify_evidence_integrity(task.task_id):
            return None
        return dict(row)

    def get_task_snapshot(self, task_id: str) -> TaskAuthoritySnapshot:
        task = self.store.get_task(task_id)
        if not task.owner_principal_id or not task.mission_id:
            raise ForbiddenError("task lacks server-owned lineage authority binding")
        proof = self._terminal_proof(task)
        return TaskAuthoritySnapshot(
            task_id=task.task_id,
            revision=task.revision,
            status=task.status,
            owner_principal_id=task.owner_principal_id,
            mission_id=task.mission_id,
            created_at=datetime.fromtimestamp(task.created_at, tz=timezone.utc),
            metadata=dict(task.task_plan.metadata),
            evidence_refs=(
                tuple(json.loads(proof["evidence_refs_json"])) if proof else ()
            ),
            terminal_evidence_proven=proof is not None,
        )

    def list_descendant_snapshots(
        self, root_task_id: str
    ) -> tuple[TaskAuthoritySnapshot, ...]:
        with self.store._lock:
            rows = self.store.conn.execute(
                "SELECT task_id FROM tasks WHERE root_task_id=? AND task_id<>? "
                "ORDER BY created_at,task_id",
                (root_task_id, root_task_id),
            ).fetchall()
        return tuple(self.get_task_snapshot(row["task_id"]) for row in rows)

    def reserve_sibling_sequence(
        self, parent_task_id: str, relation_type: RelationType, child_task_id: str
    ) -> int:
        with self.store._transaction():
            existing = self.store.conn.execute(
                "SELECT parent_task_id,relation_type,sibling_sequence "
                "FROM task_lineage_sequences WHERE child_task_id=?",
                (child_task_id,),
            ).fetchone()
            if existing is not None:
                if (
                    existing["parent_task_id"] != parent_task_id
                    or existing["relation_type"] != relation_type.value
                ):
                    raise ConflictError("child sibling reservation conflicts with lineage")
                return int(existing["sibling_sequence"])
            row = self.store.conn.execute(
                "SELECT COALESCE(MAX(sibling_sequence),0)+1 AS next_sequence "
                "FROM task_lineage_sequences WHERE parent_task_id=? AND relation_type=?",
                (parent_task_id, relation_type.value),
            ).fetchone()
            sequence = int(row["next_sequence"])
            if sequence > 99:
                raise ConflictError("lineage sibling sequence exhausted")
            self.store.conn.execute(
                "INSERT INTO task_lineage_sequences "
                "(parent_task_id,relation_type,child_task_id,sibling_sequence,created_at) "
                "VALUES(?,?,?,?,?)",
                (parent_task_id, relation_type.value, child_task_id, sequence, datetime.now(tz=timezone.utc).timestamp()),
            )
            return sequence

    def compare_and_swap_task_metadata(
        self, task_id: str, expected_revision: int, metadata_patch: Mapping[str, object]
    ) -> bool:
        if not set(metadata_patch).issubset(RESERVED_LINEAGE_METADATA_KEYS):
            raise ForbiddenError("lineage CAS contains non-lineage metadata")
        with self.store._transaction():
            task = self.store.get_task(task_id)
            if task.revision != expected_revision:
                return False
            lineage_value = metadata_patch.get(LINEAGE_METADATA_KEY)
            if isinstance(lineage_value, Mapping):
                lineage = lineage_from_metadata(lineage_value)
                if lineage.relation_type != RelationType.ROOT:
                    # The root check and child metadata CAS share this BEGIN
                    # IMMEDIATE transaction, closing the legacy attach race.
                    self.store.assert_root_open_for_child(lineage.root_task_id)
            if task.status in PROVEN_TERMINAL_STATES:
                raise ConflictError("terminal task lineage metadata is immutable")
            payload = task.task_plan.to_dict(redact=False)
            metadata = dict(payload.get("metadata") or {})
            for key, value in metadata_patch.items():
                if key in metadata and metadata[key] != value:
                    raise ConflictError("authoritative lineage metadata conflict")
                metadata[key] = value
            payload["metadata"] = metadata
            now = datetime.now(tz=timezone.utc).timestamp()
            updated = self.store.conn.execute(
                "UPDATE tasks SET task_json=?,revision=revision+1,updated_at=? "
                "WHERE task_id=? AND revision=?",
                (json.dumps(payload, ensure_ascii=False), now, task_id, expected_revision),
            )
            if updated.rowcount != 1:
                return False
            self.store.append_event(
                task_id,
                "TASK_LINEAGE_BOUND",
                {"metadata_keys": sorted(metadata_patch)},
                subject=f"task/{task_id}",
            )
            return True

    def record_terminal_proof(
        self,
        *,
        task_id: str,
        claim_id: str,
        result_digest: str,
        evidence_refs: Sequence[str],
    ) -> None:
        with self.store._transaction():
            task = self.store.get_task(task_id)
            if task.status not in PROVEN_TERMINAL_STATES:
                raise ConflictError("terminal proof requires a final task status")
            claim = self.store.conn.execute(
                "SELECT status,result_payload FROM claims WHERE claim_id=? AND task_id=?",
                (claim_id, task_id),
            ).fetchone()
            if claim is None or claim["status"] != "COMPLETED" or not claim["result_payload"]:
                raise ConflictError("terminal proof requires a completed claim")
            stored_result = json.loads(claim["result_payload"])
            if stored_result.get("result_digest") != result_digest:
                raise ConflictError("terminal proof result digest mismatch")
            refs = tuple(str(value) for value in evidence_refs)
            evidence_ids = {entry["evidence_id"] for entry in self.store.list_evidence(task_id)}
            if not set(refs).issubset(evidence_ids):
                raise ConflictError("terminal proof evidence reference mismatch")
            existing = self.store.conn.execute(
                "SELECT * FROM task_terminal_proofs WHERE task_id=?", (task_id,)
            ).fetchone()
            expected = (claim_id, result_digest, canonical_json(list(refs)))
            if existing is not None:
                actual = (
                    existing["claim_id"],
                    existing["result_digest"],
                    canonical_json(json.loads(existing["evidence_refs_json"])),
                )
                if actual != expected:
                    raise ConflictError("terminal proof replay conflict")
                return
            if not self.store.verify_event_chain(task_id) or not self.store.verify_evidence_integrity(task_id):
                raise ConflictError("terminal proof journal integrity failed")
            self.store.conn.execute(
                "INSERT INTO task_terminal_proofs "
                "(task_id,claim_id,task_revision,result_digest,evidence_refs_json,created_at) "
                "VALUES(?,?,?,?,?,?)",
                (
                    task_id,
                    claim_id,
                    task.revision,
                    result_digest,
                    json.dumps(list(refs), ensure_ascii=False),
                    datetime.now(tz=timezone.utc).timestamp(),
                ),
            )

    def get_root_review_record(self, review_id: str) -> Mapping[str, object] | None:
        row = self.store.conn.execute(
            "SELECT record_json FROM task_root_reviews WHERE review_id=?", (review_id,)
        ).fetchone()
        return None if row is None else json.loads(row["record_json"])

    def get_root_review_for_trigger(
        self, trigger_child_task_id: str
    ) -> Mapping[str, object] | None:
        row = self.store.conn.execute(
            "SELECT record_json FROM task_root_reviews "
            "WHERE trigger_child_task_id=? ORDER BY created_at LIMIT 1",
            (trigger_child_task_id,),
        ).fetchone()
        return None if row is None else json.loads(row["record_json"])

    def compare_and_swap_root_review_record(
        self,
        root_task_id: str,
        expected_root_revision: int,
        review_record: Mapping[str, object],
    ) -> bool:
        with self.store._transaction():
            review_id = str(review_record.get("review_id") or "")
            trigger_id = str(review_record.get("trigger_child_task_id") or "")
            if (
                review_record.get("root_task_id") != root_task_id
                or int(review_record.get("root_revision_before", -1)) != expected_root_revision
                or int(review_record.get("root_revision_after", -1)) != expected_root_revision + 1
            ):
                raise ConflictError("root review authority binding mismatch")
            existing = self.get_root_review_record(review_id)
            if existing is not None:
                if canonical_json(existing) != canonical_json(dict(review_record)):
                    raise ConflictError("root review replay conflict")
                return True
            trigger = self.get_task_snapshot(trigger_id)
            if not trigger.terminal_evidence_proven:
                raise ConflictError("root review trigger is not proven terminal")
            trigger_revision = trigger.revision
            frozen = self.store.conn.execute(
                "SELECT review_id FROM task_root_reviews "
                "WHERE trigger_child_task_id=? AND trigger_child_revision=?",
                (trigger_id, trigger_revision),
            ).fetchone()
            if frozen is not None:
                raise ConflictError("terminal trigger already has a different root review")
            root = self.store.get_task(root_task_id)
            if root.revision != expected_root_revision:
                return False
            now = datetime.now(tz=timezone.utc).timestamp()
            self.store.conn.execute(
                "INSERT INTO task_root_reviews "
                "(review_id,root_task_id,trigger_child_task_id,trigger_child_revision,"
                "root_revision_before,root_revision_after,record_json,created_at) "
                "VALUES(?,?,?,?,?,?,?,?)",
                (
                    review_id,
                    root_task_id,
                    trigger_id,
                    trigger_revision,
                    expected_root_revision,
                    expected_root_revision + 1,
                    json.dumps(dict(review_record), ensure_ascii=False, sort_keys=True),
                    now,
                ),
            )
            updated = self.store.conn.execute(
                "UPDATE tasks SET revision=revision+1,updated_at=? "
                "WHERE task_id=? AND revision=?",
                (now, root_task_id, expected_root_revision),
            )
            if updated.rowcount != 1:
                raise ConflictError("root review CAS lost after reservation")
            self.store.append_event(
                root_task_id,
                "ROOT_REVIEW",
                {
                    "review_id": review_id,
                    "trigger_child_task_id": trigger_id,
                    "next_legal_action": review_record.get("next_legal_action"),
                    "root_complete": bool(review_record.get("root_complete")),
                },
                subject=f"task/{root_task_id}",
            )
            return True

    def get_root_action_dispatch(self, review_id: str) -> RootActionDispatchReceipt | None:
        row = self.store.conn.execute(
            "SELECT * FROM task_root_action_dispatches WHERE review_id=?", (review_id,)
        ).fetchone()
        if row is None:
            return None
        return RootActionDispatchReceipt(
            review_id=row["review_id"],
            root_task_id=row["root_task_id"],
            action=NextLegalAction(row["action"]),
            idempotency_key=row["idempotency_key"],
            status=row["status"],
            mission_id=row["mission_id"],
            root_revision=int(row["root_revision"]),
            plan_digest=row["plan_digest"],
            execution_key=row["execution_key"],
        )

    def dispatch_root_action(
        self, review: RootReviewRecord, idempotency_key: str
    ) -> RootActionDispatchReceipt:
        with self.store._transaction():
            persisted = self.get_root_review_record(review.review_id)
            if persisted is None:
                raise ConflictError("root action requires a persisted root review")
            existing = self.get_root_action_dispatch(review.review_id)
            if existing is not None:
                if (
                    existing.root_task_id != review.root_task_id
                    or existing.action != review.next_legal_action
                    or existing.idempotency_key != idempotency_key
                ):
                    raise ConflictError("root action dispatch replay conflict")
                return existing
            now = datetime.now(tz=timezone.utc).timestamp()
            self.store.conn.execute(
                "INSERT INTO task_root_action_dispatches "
                "(review_id,root_task_id,action,idempotency_key,status,created_at,"
                "mission_id,root_revision,updated_at) VALUES(?,?,?,?,?,?,?,?,?)",
                (
                    review.review_id,
                    review.root_task_id,
                    review.next_legal_action.value,
                    idempotency_key,
                    "PENDING",
                    now,
                    self.store.get_task(review.root_task_id).mission_id,
                    review.root_revision_after,
                    now,
                ),
            )
            self.store.append_event(
                review.root_task_id,
                "ROOT_NEXT_ACTION_PENDING",
                {
                    "review_id": review.review_id,
                    "action": review.next_legal_action.value,
                    "idempotency_key": idempotency_key,
                },
                subject=f"task/{review.root_task_id}",
            )
            return self.get_root_action_dispatch(review.review_id)  # type: ignore[return-value]


class TaskPlaneLineageCoordinator:
    """Coordinates lineage with real TaskPlane admission and completion paths."""

    def __init__(self, store: TaskPlaneStore, policy: TaskLineagePolicy) -> None:
        self.store = store
        self.policy = policy
        self.binding = TaskPlaneLineageStoreBinding(store)
        self.adapter = InProcessTaskPlaneLineageAdapter(
            self.binding.service_bindings(), readback=self.binding
        )
        self.service = TaskIdentityLineageService(self.adapter)
        self.hooks = TaskLineageLifecycleHooks(self.adapter)

    @staticmethod
    def _request_digest(payload: Mapping[str, object]) -> str:
        return compute_sha256_hex(dict(payload))

    @staticmethod
    def _reject_reserved_metadata(plan: TaskPlan) -> None:
        if set(plan.metadata).intersection(RESERVED_LINEAGE_METADATA_KEYS):
            raise ValidationError("lineage metadata is server-owned")

    @staticmethod
    def _goal_slug(plan: TaskPlan) -> str:
        for candidate in (plan.intent, plan.task_type, f"TASK-{plan.task_id}"):
            if not candidate:
                continue
            try:
                return normalize_goal_slug(str(candidate))
            except ValueError:
                continue
        raise ValidationError("task goal cannot be normalized for lineage")

    def _existing_by_idempotency(self, key: str | None) -> TaskRecord | None:
        if not key:
            return None
        row = self.store.conn.execute(
            "SELECT task_id FROM tasks WHERE idempotency_key=?", (key,)
        ).fetchone()
        return None if row is None else self.store.get_task(row["task_id"])

    @staticmethod
    def _verify_replay(
        task: TaskRecord,
        *,
        request_digest: str,
        principal: PrincipalLineageBinding,
        parent_task_id: str | None,
    ) -> TaskRecord:
        metadata = task.task_plan.metadata
        if (
            metadata.get(LINEAGE_REQUEST_DIGEST_KEY) != request_digest
            or task.owner_principal_id != principal.owner_principal_id
            or task.mission_id != principal.mission_id
            or task.task_plan.parent_task_id != parent_task_id
        ):
            raise ConflictError("lineage admission replay conflict")
        return task

    def create_root(
        self,
        *,
        plan: TaskPlan,
        authenticated_identity: str,
        request_digest: str,
        idempotency_key: str | None,
    ) -> TaskRecord:
        principal = self.policy.resolve_principal(authenticated_identity)
        return self.create_root_for_principal(
            plan=plan,
            principal=principal,
            request_digest=request_digest,
            idempotency_key=idempotency_key,
        )

    def create_root_for_principal(
        self,
        *,
        plan: TaskPlan,
        principal: PrincipalLineageBinding,
        request_digest: str,
        idempotency_key: str | None,
        created_at: datetime | None = None,
    ) -> TaskRecord:
        self.policy.assert_acceptance_ref_registered(principal.root_acceptance_ref)
        self._reject_reserved_metadata(plan)
        with self.store._transaction():
            existing = self._existing_by_idempotency(idempotency_key)
            if existing is not None:
                return self._verify_replay(
                    existing,
                    request_digest=request_digest,
                    principal=principal,
                    parent_task_id=None,
                )
            created_at = created_at or datetime.now(tz=timezone.utc)
            if created_at.tzinfo is None or created_at.utcoffset() is None:
                raise ValidationError("lineage created_at must be timezone-aware")
            created_at = created_at.astimezone(timezone.utc)
            prepared = self.service.prepare_new_task_identity(
                NewTaskIdentityRequest(
                    task_id=str(plan.task_id),
                    owner_principal_id=principal.owner_principal_id,
                    preferred_owner_alias=principal.preferred_owner_alias,
                    created_at=created_at,
                    mission_id=principal.mission_id,
                    goal_slug=self._goal_slug(plan),
                    root_acceptance_ref=principal.root_acceptance_ref,
                    relation_type=RelationType.ROOT,
                )
            )
            plan.metadata = {
                **plan.metadata,
                **prepared.metadata_patch,
                LINEAGE_REQUEST_DIGEST_KEY: request_digest,
            }
            return self.store.create_task_with_event(
                plan,
                idempotency_key=idempotency_key,
                owner_principal_id=principal.owner_principal_id,
                mission_id=principal.mission_id,
                created_at=created_at.timestamp(),
            )

    def delegate_child(
        self,
        *,
        parent_task_id: str,
        plan: TaskPlan,
        authenticated_identity: str,
        request_digest: str,
        delegation: DelegationEngine,
        principal: PrincipalLineageBinding | None = None,
    ) -> TaskRecord:
        if not plan.task_id:
            raise ValidationError("lineage delegation requires a stable task_id")
        principal = principal or self.policy.resolve_principal(authenticated_identity)
        if principal.authenticated_identity != authenticated_identity:
            raise ForbiddenError("delegation principal identity mismatch")
        self._reject_reserved_metadata(plan)
        with self.store._transaction():
            parent = self.store.get_task(parent_task_id)
            root = self.store.get_task(parent.root_task_id)
            if (
                root.owner_principal_id != principal.owner_principal_id
                or root.mission_id != principal.mission_id
            ):
                raise ForbiddenError("delegation actor does not match root authority")
            try:
                existing = self.store.get_task(str(plan.task_id))
            except NotFoundError:
                existing = None
            if existing is not None:
                return self._verify_replay(
                    existing,
                    request_digest=request_digest,
                    principal=principal,
                    parent_task_id=parent_task_id,
                )
            self.store.assert_root_open_for_child(root.task_id)
            created_at = datetime.now(tz=timezone.utc)
            prepared = self.service.prepare_new_task_identity(
                NewTaskIdentityRequest(
                    task_id=str(plan.task_id),
                    owner_principal_id=root.owner_principal_id,
                    preferred_owner_alias=principal.preferred_owner_alias,
                    created_at=created_at,
                    mission_id=root.mission_id,
                    goal_slug=self._goal_slug(plan),
                    root_acceptance_ref=str(
                        root.task_plan.metadata[LINEAGE_METADATA_KEY]["root_acceptance_ref"]
                    ),
                    relation_type=RelationType.CHILD,
                    parent_task_id=parent_task_id,
                )
            )
            plan.metadata = {
                **plan.metadata,
                **prepared.metadata_patch,
                LINEAGE_REQUEST_DIGEST_KEY: request_digest,
            }
            return delegation.delegate(
                parent_task_id,
                plan,
                caller=authenticated_identity,
                owner_principal_id=root.owner_principal_id,
                mission_id=root.mission_id,
                created_at=created_at.timestamp(),
            )

    def current_acceptance_snapshot(self, *, root_task_id: str) -> RootAcceptanceSnapshot:
        """Evaluate the current authoritative root/descendant projection."""
        root = _as_task_record(self.binding.get_task_snapshot(root_task_id))
        descendants = tuple(
            _as_task_record(snapshot)
            for snapshot in self.binding.list_descendant_snapshots(root_task_id)
        )
        return self.policy.evaluate_acceptance(root, descendants)

    def review_proven_terminal_child(
        self,
        *,
        child_task_id: str,
        claim_id: str,
        result_digest: str,
        evidence_refs: Sequence[str],
    ):
        with self.store._transaction():
            self.binding.record_terminal_proof(
                task_id=child_task_id,
                claim_id=claim_id,
                result_digest=result_digest,
                evidence_refs=evidence_refs,
            )
            return self.resume_terminal_review(child_task_id=child_task_id)

    def resume_terminal_review(self, *, child_task_id: str) -> TerminalReviewOutcome:
        """Resume after restart without re-completing a claim or re-evaluating a review.

        Once a trigger has a persisted review, that exact decision is frozen.
        Policy/evaluator drift therefore cannot create a second decision for the
        same terminal transition.
        """
        with self.store._transaction():
            child = self.binding.get_task_snapshot(child_task_id)
            if not child.terminal_evidence_proven:
                raise ConflictError("terminal review resume requires durable proof")
            raw_lineage = child.metadata.get(LINEAGE_METADATA_KEY)
            if not isinstance(raw_lineage, Mapping):
                raise ConflictError("lineage child lacks authoritative lineage metadata")
            try:
                child_lineage = lineage_from_metadata(dict(raw_lineage))
            except ValueError as exc:
                raise ConflictError("lineage child metadata is invalid") from exc
            persisted = self.binding.get_root_review_for_trigger(child_task_id)
            if persisted is not None:
                review_id = persisted.get("review_id")
                if not isinstance(review_id, str) or not review_id:
                    raise ConflictError("persisted root review identity is invalid")
                try:
                    review = root_review_from_record(
                        persisted, expected_review_id=review_id
                    )
                except ValueError as exc:
                    raise ConflictError("persisted root review is invalid") from exc
                root = self.binding.get_task_snapshot(child_lineage.root_task_id)
                if (
                    review.trigger_child_task_id != child_task_id
                    or review.root_task_id != child_lineage.root_task_id
                    or child_lineage.task_id != child_task_id
                    or child_lineage.owner_principal_id != child.owner_principal_id
                    or child_lineage.mission_id != child.mission_id
                    or root.owner_principal_id != child.owner_principal_id
                    or root.mission_id != child.mission_id
                    or root.revision < review.root_revision_after
                ):
                    raise ConflictError("persisted root review authority mismatch")
                dispatch = self.binding.get_root_action_dispatch(review.review_id)
                if dispatch is None:
                    dispatch = self.binding.dispatch_root_action(
                        review,
                        f"root-review:{review.review_id}:{review.next_legal_action.value}",
                    )
                expected_key = (
                    f"root-review:{review.review_id}:{review.next_legal_action.value}"
                )
                if (
                    dispatch.review_id != review.review_id
                    or dispatch.root_task_id != review.root_task_id
                    or dispatch.action != review.next_legal_action
                    or dispatch.idempotency_key != expected_key
                    or dispatch.status not in PERSISTED_DISPATCH_STATUSES
                ):
                    raise ConflictError("persisted root action dispatch is invalid")
                return TerminalReviewOutcome(review=review, dispatch=dispatch)
            hook_outcome = self.hooks.on_proven_terminal_child(
                child_task_id=child_task_id,
                context=AuthenticatedTaskContext(
                    principal_id=child.owner_principal_id,
                    preferred_alias=child_lineage.owner_alias,
                    mission_id=child.mission_id,
                ),
                acceptance_evaluator=self.policy.evaluate_acceptance,
                reviewed_at=datetime.now(tz=timezone.utc),
                require_independent_readback=False,
            )
            return hook_outcome.outcome