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

"""Mission-owned, append-only bindings for verified external legacy executions."""
from __future__ import annotations

import json
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Mapping, Protocol, Sequence

from .errors import ConflictError, NotFoundError, ValidationError
from .models import compute_sha256_hex


ATTACH_REQUEST_SCHEMA = "aios.mission-legacy-attach-request.v1"
ATTACH_EVIDENCE_SCHEMA = "aios.verified-legacy-attachment-evidence.v1"
ATTACH_RECEIPT_SCHEMA = "aios.mission-legacy-attachment-receipt.v1"
ATTACHMENT_KIND = "LEGACY_EXECUTION_SUBJECT"
RELATION_TYPE = "RECOVERY"
SOURCE_KIND = "agent-task-bridge-job"
REQUIRED_READBACK_ROLES = frozenset(
    {"agent_task", "admission", "owner_registry", "bridge_job"}
)
REQUIRED_C04_EVIDENCE = frozenset(
    {
        "verified legacy ownership and acceptance relevance",
        "unchanged task/job/idempotency/reconciliation identity",
        "authoritative attach readback",
    }
)
_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")


def _canonical(value: object) -> str:
    try:
        return json.dumps(
            value,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        )
    except (TypeError, ValueError) as exc:
        raise ValidationError("legacy attachment value must be JSON data") from exc


def _object(value: object, fields: frozenset[str], name: str) -> dict[str, object]:
    if not isinstance(value, Mapping) or set(value) != fields:
        raise ValidationError(f"invalid {name} fields")
    return json.loads(_canonical(dict(value)))


def _identifier(value: object, name: str, *, maximum: int = 256) -> str:
    if (
        not isinstance(value, str)
        or not value
        or len(value) > maximum
        or any(ord(character) < 33 for character in value)
    ):
        raise ValidationError(f"invalid {name}")
    return value


def _integer(value: object, name: str) -> int:
    if type(value) is not int or value < 0:
        raise ValidationError(f"invalid {name}")
    return value


def _utc(value: object, name: str) -> str:
    if not isinstance(value, str) or not value:
        raise ValidationError(f"invalid {name}")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ValidationError(f"invalid {name}") from exc
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise ValidationError(f"invalid {name}")
    return parsed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")


@dataclass(frozen=True)
class AuthorityReadbackRef:
    """A non-secret pointer and digest issued by one independent authority."""

    role: str
    authority: str
    readback_ref: str
    canonical_sha256: str
    observed_at: str
    revision: int | None = None

    def to_dict(self) -> dict[str, object]:
        if self.role not in REQUIRED_READBACK_ROLES:
            raise ValidationError("invalid legacy authority readback role")
        authority = _identifier(self.authority, "readback authority")
        reference = _identifier(self.readback_ref, "readback ref", maximum=1024)
        if not _SHA256_RE.fullmatch(self.canonical_sha256):
            raise ValidationError("invalid readback canonical SHA-256")
        observed = _utc(self.observed_at, "readback observed_at")
        revision = self.revision
        if revision is not None:
            revision = _integer(revision, "readback revision")
        return {
            "role": self.role,
            "authority": authority,
            "readback_ref": reference,
            "canonical_sha256": self.canonical_sha256,
            "observed_at": observed,
            "revision": revision,
        }


@dataclass(frozen=True)
class VerifiedLegacyAttachmentEvidence:
    """Typed output of a configured authority reader, never a caller payload."""

    schema: str
    source_kind: str
    source_task_id: str
    source_task_created_at: str
    source_task_status: str
    source_task_revision: int
    source_task_attempt: int
    source_task_max_attempts: int
    external_job_id: str
    internal_job_id: str
    internal_idempotency_key: str
    job_branch: str
    job_state: str
    job_revision: int
    side_effect_state: str
    grant_id: str
    grant_digest: str
    execution_digest: str
    epoch_id: str
    epoch_digest: str
    historical_signer_id: str
    historical_signer_fingerprint: str
    registry_generation: int
    signature_verified: bool
    epoch_currently_valid: bool
    readbacks: Sequence[AuthorityReadbackRef]

    def to_dict(self) -> dict[str, object]:
        if self.schema != ATTACH_EVIDENCE_SCHEMA or self.source_kind != SOURCE_KIND:
            raise ValidationError("invalid legacy attachment evidence schema")
        if self.signature_verified is not True or self.epoch_currently_valid is not False:
            raise ConflictError("legacy historical Owner evidence is not verified-expired")
        values = {
            "source_task_id": self.source_task_id,
            "source_task_status": self.source_task_status,
            "external_job_id": self.external_job_id,
            "internal_job_id": self.internal_job_id,
            "internal_idempotency_key": self.internal_idempotency_key,
            "job_branch": self.job_branch,
            "job_state": self.job_state,
            "side_effect_state": self.side_effect_state,
            "grant_id": self.grant_id,
            "historical_signer_id": self.historical_signer_id,
            "historical_signer_fingerprint": self.historical_signer_fingerprint,
            "epoch_id": self.epoch_id,
        }
        normalized = {
            key: _identifier(value, key, maximum=1024)
            for key, value in values.items()
        }
        for name, value in {
            "grant_digest": self.grant_digest,
            "execution_digest": self.execution_digest,
            "epoch_digest": self.epoch_digest,
        }.items():
            if not _SHA256_RE.fullmatch(value):
                raise ValidationError(f"invalid {name}")
            normalized[name] = value
        task_revision = _integer(self.source_task_revision, "source_task_revision")
        task_attempt = _integer(self.source_task_attempt, "source_task_attempt")
        task_max_attempts = _integer(
            self.source_task_max_attempts, "source_task_max_attempts"
        )
        if task_attempt > task_max_attempts:
            raise ConflictError("legacy task attempt exceeds max attempts")
        job_revision = _integer(self.job_revision, "job_revision")
        registry_generation = _integer(
            self.registry_generation, "registry_generation"
        )
        readbacks = [value.to_dict() for value in self.readbacks]
        roles = [str(value["role"]) for value in readbacks]
        if len(roles) != len(set(roles)) or set(roles) != REQUIRED_READBACK_ROLES:
            raise ValidationError("legacy evidence requires four authority readbacks")
        readbacks.sort(key=lambda value: str(value["role"]))
        return {
            "schema": ATTACH_EVIDENCE_SCHEMA,
            "source_kind": SOURCE_KIND,
            **normalized,
            "source_task_created_at": _utc(
                self.source_task_created_at, "source_task_created_at"
            ),
            "source_task_revision": task_revision,
            "source_task_attempt": task_attempt,
            "source_task_max_attempts": task_max_attempts,
            "job_revision": job_revision,
            "registry_generation": registry_generation,
            "signature_verified": True,
            "epoch_currently_valid": False,
            "readbacks": readbacks,
        }


class LegacyAttachmentEvidenceProvider(Protocol):
    """Configured server-side reader for the four independent authority sources."""

    def read_verified(
        self,
        *,
        mission_id: str,
        root_task_id: str,
        anchor_task_id: str,
        source_kind: str,
        source_task_id: str,
        external_job_id: str,
        original_acceptance_digest: str,
    ) -> VerifiedLegacyAttachmentEvidence: ...


def normalize_attach_request(value: object) -> dict[str, object]:
    body = _object(
        value,
        frozenset({"schema", "anchor_task_id", "source", "expected"}),
        "legacy attachment request",
    )
    if body["schema"] != ATTACH_REQUEST_SCHEMA:
        raise ValidationError("invalid legacy attachment request schema")
    source = _object(
        body["source"],
        frozenset({"kind", "task_id", "external_job_id"}),
        "legacy attachment source",
    )
    if source["kind"] != SOURCE_KIND:
        raise ValidationError("invalid legacy attachment source kind")
    expected = _object(
        body["expected"],
        frozenset(
            {
                "root_revision",
                "anchor_revision",
                "agent_task_revision",
                "agent_task_attempt",
                "internal_job_id",
                "job_revision",
                "internal_idempotency_key",
            }
        ),
        "legacy attachment expected",
    )
    return {
        "schema": ATTACH_REQUEST_SCHEMA,
        "anchor_task_id": _identifier(body["anchor_task_id"], "anchor_task_id"),
        "source": {
            "kind": SOURCE_KIND,
            "task_id": _identifier(source["task_id"], "source task_id"),
            "external_job_id": _identifier(
                source["external_job_id"], "external_job_id"
            ),
        },
        "expected": {
            "root_revision": _integer(expected["root_revision"], "root_revision"),
            "anchor_revision": _integer(
                expected["anchor_revision"], "anchor_revision"
            ),
            "agent_task_revision": _integer(
                expected["agent_task_revision"], "agent_task_revision"
            ),
            "agent_task_attempt": _integer(
                expected["agent_task_attempt"], "agent_task_attempt"
            ),
            "internal_job_id": _identifier(
                expected["internal_job_id"], "internal_job_id"
            ),
            "job_revision": _integer(expected["job_revision"], "job_revision"),
            "internal_idempotency_key": _identifier(
                expected["internal_idempotency_key"], "internal_idempotency_key"
            ),
        },
    }


def validate_c04_acceptance(
    acceptance: Mapping[str, object], *, source_task_id: str, external_job_id: str
) -> None:
    candidate = acceptance.get("legacy_candidate")
    if (
        not isinstance(candidate, Mapping)
        or candidate.get("task_id") != source_task_id
        or candidate.get("job_id") != external_job_id
    ):
        raise ConflictError("mission legacy candidate authority mismatch")
    conditions = acceptance.get("conditions")
    if not isinstance(conditions, list):
        raise ConflictError("mission C04 acceptance condition is unavailable")
    matches = [
        condition
        for condition in conditions
        if isinstance(condition, Mapping) and condition.get("key") == "C04"
    ]
    if len(matches) != 1:
        raise ConflictError("mission C04 acceptance condition is unavailable")
    condition = matches[0]
    evidence = condition.get("required_evidence")
    if (
        condition.get("purpose") != "LEGACY-LINEAGE-MIGRATION"
        or not isinstance(evidence, list)
        or not REQUIRED_C04_EVIDENCE.issubset(
            value for value in evidence if isinstance(value, str)
        )
    ):
        raise ConflictError("mission C04 acceptance contract mismatch")


def validate_evidence_matches_request(
    evidence: Mapping[str, object], request: Mapping[str, object]
) -> None:
    source = request["source"]
    expected = request["expected"]
    assert isinstance(source, Mapping) and isinstance(expected, Mapping)
    pairs = {
        "source_kind": source["kind"],
        "source_task_id": source["task_id"],
        "external_job_id": source["external_job_id"],
        "source_task_revision": expected["agent_task_revision"],
        "source_task_attempt": expected["agent_task_attempt"],
        "internal_job_id": expected["internal_job_id"],
        "job_revision": expected["job_revision"],
        "internal_idempotency_key": expected["internal_idempotency_key"],
    }
    mismatches = [name for name, value in pairs.items() if evidence.get(name) != value]
    if mismatches:
        raise ConflictError(
            "legacy provider evidence mismatch: " + ",".join(sorted(mismatches))
        )


class LegacyAttachmentLedger:
    """Append-only persistence owned by one Mission authority store."""

    def __init__(self, store) -> None:
        self.store = store
        with store._transaction():
            store.conn.execute(
                """
                CREATE TABLE IF NOT EXISTS task_mission_legacy_attachments (
                    attachment_id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    root_task_id TEXT NOT NULL,
                    anchor_task_id TEXT NOT NULL,
                    authenticated_identity TEXT NOT NULL,
                    owner_principal_id TEXT NOT NULL,
                    attachment_kind TEXT NOT NULL,
                    relation_type TEXT NOT NULL,
                    attachment_sequence INTEGER NOT NULL,
                    display_name TEXT NOT NULL,
                    source_kind TEXT NOT NULL,
                    source_task_id TEXT NOT NULL,
                    external_job_id TEXT NOT NULL,
                    internal_job_id TEXT NOT NULL,
                    internal_idempotency_key TEXT NOT NULL,
                    request_digest TEXT NOT NULL,
                    idempotency_key TEXT NOT NULL,
                    evidence_digest TEXT NOT NULL,
                    receipt_digest TEXT NOT NULL,
                    receipt_json TEXT NOT NULL,
                    created_at REAL NOT NULL,
                    UNIQUE (authenticated_identity, idempotency_key),
                    UNIQUE (source_kind, source_task_id),
                    UNIQUE (source_kind, external_job_id),
                    UNIQUE (mission_id, anchor_task_id, attachment_sequence),
                    UNIQUE (mission_id, anchor_task_id, source_kind, source_task_id, external_job_id)
                )
                """
            )
            store.conn.execute(
                "CREATE INDEX IF NOT EXISTS ix_mission_legacy_attachments "
                "ON task_mission_legacy_attachments(mission_id, anchor_task_id)"
            )
            store.conn.execute(
                """
                CREATE TABLE IF NOT EXISTS task_mission_acceptance_child_bindings (
                    mission_id TEXT NOT NULL,
                    condition_key TEXT NOT NULL,
                    child_task_id TEXT NOT NULL UNIQUE,
                    original_acceptance_digest TEXT NOT NULL,
                    binding_digest TEXT NOT NULL,
                    created_at REAL NOT NULL,
                    PRIMARY KEY (mission_id, condition_key)
                )
                """
            )

    def acceptance_child_binding(self, mission_id: str, condition_key: str):
        return self.store.conn.execute(
            "SELECT * FROM task_mission_acceptance_child_bindings "
            "WHERE mission_id=? AND condition_key=?",
            (mission_id, condition_key),
        ).fetchone()

    def bind_acceptance_child(
        self,
        *,
        mission_id: str,
        condition_key: str,
        child_task_id: str,
        original_acceptance_digest: str,
        binding_digest: str,
        created_at: float,
    ) -> None:
        existing = self.acceptance_child_binding(mission_id, condition_key)
        if existing is not None:
            if (
                existing["child_task_id"] != child_task_id
                or existing["original_acceptance_digest"]
                != original_acceptance_digest
                or existing["binding_digest"] != binding_digest
            ):
                raise ConflictError(
                    "mission acceptance condition already has a canonical child"
                )
            return
        child_binding = self.store.conn.execute(
            "SELECT mission_id,condition_key FROM "
            "task_mission_acceptance_child_bindings WHERE child_task_id=?",
            (child_task_id,),
        ).fetchone()
        if child_binding is not None:
            raise ConflictError("mission child already belongs to an acceptance condition")
        self.store.conn.execute(
            "INSERT INTO task_mission_acceptance_child_bindings "
            "(mission_id,condition_key,child_task_id,original_acceptance_digest,"
            "binding_digest,created_at) VALUES(?,?,?,?,?,?)",
            (
                mission_id,
                condition_key,
                child_task_id,
                original_acceptance_digest,
                binding_digest,
                created_at,
            ),
        )

    def by_idempotency(self, identity: str, key: str):
        return self.store.conn.execute(
            "SELECT * FROM task_mission_legacy_attachments "
            "WHERE authenticated_identity=? AND idempotency_key=?",
            (identity, key),
        ).fetchone()

    def by_source(self, source_kind: str, task_id: str, external_job_id: str):
        return self.store.conn.execute(
            "SELECT * FROM task_mission_legacy_attachments "
            "WHERE source_kind=? AND (source_task_id=? OR external_job_id=?)",
            (source_kind, task_id, external_job_id),
        ).fetchone()

    def by_id(self, mission_id: str, attachment_id: str):
        row = self.store.conn.execute(
            "SELECT * FROM task_mission_legacy_attachments "
            "WHERE mission_id=? AND attachment_id=?",
            (mission_id, attachment_id),
        ).fetchone()
        if row is None:
            raise NotFoundError("legacy attachment not found")
        return row

    def next_sequence(self, mission_id: str, anchor_task_id: str) -> int:
        row = self.store.conn.execute(
            "SELECT COALESCE(MAX(attachment_sequence),0)+1 "
            "FROM task_mission_legacy_attachments "
            "WHERE mission_id=? AND anchor_task_id=?",
            (mission_id, anchor_task_id),
        ).fetchone()
        sequence = int(row[0])
        if not 1 <= sequence <= 99:
            raise ConflictError("legacy attachment sequence exhausted")
        return sequence

    def insert(
        self,
        *,
        receipt: Mapping[str, object],
        authenticated_identity: str,
        idempotency_key: str,
        request_digest: str,
        evidence_digest: str,
        created_at: float,
    ) -> None:
        self.store.conn.execute(
            "INSERT INTO task_mission_legacy_attachments "
            "(attachment_id,mission_id,root_task_id,anchor_task_id,"
            "authenticated_identity,owner_principal_id,attachment_kind,"
            "relation_type,attachment_sequence,display_name,source_kind,"
            "source_task_id,external_job_id,internal_job_id,"
            "internal_idempotency_key,request_digest,idempotency_key,"
            "evidence_digest,receipt_digest,receipt_json,created_at) "
            "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                receipt["attachment_id"], receipt["mission_id"],
                receipt["root_task_id"], receipt["anchor_task_id"],
                authenticated_identity, receipt["owner_principal_id"],
                receipt["attachment_kind"], receipt["relation_type"],
                receipt["attachment_sequence"], receipt["display_name"],
                receipt["source"]["source_kind"],
                receipt["source"]["source_task_id"],
                receipt["source"]["external_job_id"],
                receipt["source"]["internal_job_id"],
                receipt["source"]["internal_idempotency_key"],
                request_digest, idempotency_key, evidence_digest,
                receipt["receipt_digest"], _canonical(receipt), created_at,
            ),
        )

    @staticmethod
    def receipt(row) -> dict[str, object]:
        try:
            value = json.loads(row["receipt_json"])
        except (TypeError, ValueError, json.JSONDecodeError) as exc:
            raise ConflictError("persisted legacy attachment receipt is invalid") from exc
        if (
            not isinstance(value, dict)
            or value.get("schema") != ATTACH_RECEIPT_SCHEMA
            or value.get("receipt_digest") != row["receipt_digest"]
        ):
            raise ConflictError("persisted legacy attachment receipt is invalid")
        digest_payload = dict(value)
        digest_payload.pop("receipt_digest", None)
        digest_payload.pop("idempotent", None)
        if compute_sha256_hex(digest_payload) != row["receipt_digest"]:
            raise ConflictError("persisted legacy attachment receipt digest mismatch")
        return value