"""Execute an admitted task through its existing native and evidence owners.

One current intent fences each effect. Native dispatches, assessments and
supervision retain their original authorities; the task stores references,
not another assignment history. Waiting/reading happens outside the primary
transaction. An unknown effect never becomes a new dispatch on restart.
"""

import json
import os
from pathlib import Path
import time
from contextlib import ExitStack, contextmanager
from typing import Callable

from . import cli, engagement, load_set, progression_owner as authority
from . import progression_runtime, recovery, report_delivery, report_gates, supervision
from . import workflow_contract as schema, workflow_dispatch as dispatch
from . import workflow_planning, workflow_readiness, workflow_workspace as workspace
from . import workflow_artifacts, workflow_upgrade
from .errors import ForemanError, UsageError
from .state import TransactionBusy, save_state, state_lock
from .workflow_native import NativeWorkflowAdapter

CHECKPOINT_LOCK_BUDGET_SEC = 5
CHECKPOINT_LOCK_POLL_SEC = 0.01


def encoded(value):
    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")


class WorkflowService:
    def __init__(self, capability, client, *, clock=cli.now_iso, sleeper=time.sleep,
                 native: Callable[..., NativeWorkflowAdapter] = NativeWorkflowAdapter,
                 planner=workflow_planning.plan, release=None):
        if not isinstance(capability, authority.ServiceCapability):
            raise UsageError("Start task execution with its actual admitted service capability.", {})
        self.capability, self.client = capability, client
        self.clock, self.sleeper = clock, sleeper
        if release is None:
            from .workflow_release import WorkflowReleaseAdapter
            release = WorkflowReleaseAdapter()
        self.native, self.planner, self.release = native, planner, release
        self._effect_lock_held = False
        state = authority._load(capability.state_path)
        capability.require_current(capability.state_path, state, capability.task)
        authority.verify_foreman(capability.state_path, client, native_only=True)
        current = workflow_readiness.current_inputs(state, capability.state_path, capability.task,
            supervision.load(capability.state_path), capability.plugin_root)
        self.document, self.profile = current["document"], current["profile"]
        self.inputs, self.policy = current["inputs"], current["policy"]
        self.agents, self.backpressure, self.by_kind = current["agents"], current["backpressure"], current["by_kind"]
        capability._workflow_executor = self

    @property
    def path(self):
        return self.capability.state_path

    def _current(self):
        state = authority._load(self.path)
        self.capability.require_current(self.path, state, self.capability.task)
        task = state["recovery"]["tasks"][self.capability.task]
        progress = task.get("progression")
        schema.validate_current(progress, self.document)
        if progress["generation"] != self.capability.generation:
            raise UsageError("Restore the current task generation before resuming its workflow.", {})
        workspace.inputs(self.document)
        return state, task, progress

    def _save(self, state):
        self.capability.lease.require_current()
        authority.verify_foreman(self.path, self.client, native_only=True)
        schema.validate_current(state["recovery"]["tasks"][self.capability.task]["progression"], self.document)
        save_state(self.path, state)

    def _view(self, progress, status, reason=None, *, capacity=None):
        delivery = progress["delivery"]
        if delivery is not None and delivery["schema_version"] == 2:
            delivery = {**delivery, "artifacts": workflow_artifacts.view(delivery["artifacts"])}
        result = {"schema_version": 2, "wake_agent": status == "exception", "data": {
            "task": self.capability.task, "generation": self.capability.generation,
            "status": status, "phase": progress["phase"], "correction": progress["round"],
            "jobs": progress["jobs"], "pending_intent": progress["pending"]["id"] if progress["pending"] else None,
            "reason": reason, "exception": progress["exception"],
            "exception_digest": supervision.digest(progress["exception"]) if progress["exception"] is not None else None,
            "delivery": delivery,
            "authorizes_acceptance": False, "authorizes_release": False}}
        if capacity is not None:
            result["data"]["capacity"] = capacity
        return load_set._bounded(result)

    def _exception(self, state, code, sources=(), actions=("inspect_current_evidence", "return_authority")):
        progress = state["recovery"]["tasks"][self.capability.task]["progression"]
        phase = progress["exception"]["phase"] if progress["phase"] == "exception" else progress["phase"]
        at = self.clock()
        occurrence = supervision.digest([self.capability.task, self.capability.generation, phase, at,
                                         len(state["recovery"]["events"])])
        progress.update(phase="exception", at=at, exception={
            "id": occurrence, "code": code, "phase": phase, "sources": list(dict.fromkeys(
                "".join(char if ord(char) >= 32 else " " for char in str(item)).encode("utf-8")[:4096].decode("utf-8", errors="ignore")
                for item in sources))[:8], "actions": list(actions)})
        self._save(state)
        return self._view(progress, "exception", code)

    def _intent(self, progress, operation, *, job=None, inputs=(), dispatches=()):
        identifier = "workflow-" + supervision.digest([self.capability.task, self.capability.generation,
            progress["round"], progress["phase"], operation, job["id"] if job else None])[:48]
        from .release_activation import enabled
        argv = ["--now", self.clock()] if operation == "measure" and enabled() else []
        progress["pending"] = {"schema_version": 2, "id": identifier, "operation": operation,
            "argv": argv, "inputs": list(inputs), "effects_started": False,
            "dispatches": list(dispatches), "job": job}

    def _before_effect(self, state):
        progress = state["recovery"]["tasks"][self.capability.task]["progression"]
        intent = progress["pending"]
        if intent is None:
            raise UsageError("Restore this task's durable effect intent before native transport.", {})
        for expected in intent["inputs"]:
            if workspace.capture(expected["path"])[0] != expected:
                raise UsageError("A pending effect's input changed; preserve its original obligation and reconcile before transport.", {})
        if not intent["effects_started"]:
            intent["effects_started"] = True
            self._save(state)

    @contextmanager
    def _checkpoint_lock(self):
        """Wait boundedly for a short save, without repeating its body."""
        with ExitStack() as transaction:
            deadline = time.monotonic() + CHECKPOINT_LOCK_BUDGET_SEC
            while True:
                try:
                    transaction.enter_context(state_lock(self.path))
                    break
                except TransactionBusy:
                    if time.monotonic() >= deadline:
                        raise
                    # Retry only lock acquisition. The checkpoint body and
                    # native effect below this boundary execute at most once.
                    time.sleep(CHECKPOINT_LOCK_POLL_SEC)
            yield

    @contextmanager
    def _effect_transaction(self, state, identifier, job):
        """Refresh canonical state for one short transport checkpoint.

        Native startup/waits run outside this transaction. Another task's
        assignments may append between callbacks, so never save a stale
        whole-owner snapshot or reuse its earlier assignment indexes.
        """
        if self._effect_lock_held:
            yield
            return
        with self._checkpoint_lock():
            current, _task, progress = self._current()
            if progress["pending"] is None or progress["pending"]["id"] != identifier:
                raise UsageError("The task's pending native intent changed; preserve the original effect for its owner.", {})
            if job is not None:
                canonical = progress["pending"]["job"]
                if canonical is None or canonical["id"] != job["id"]:
                    raise UsageError("Preserve the original pending native job before saving another transport checkpoint.", {})
                job.clear()
                job.update(canonical)
                progress["pending"]["job"] = job
            state.clear()
            state.update(current)
            self._effect_lock_held = True
            try:
                yield
            finally:
                self._effect_lock_held = False

    def _roles(self, progress):
        return workflow_readiness.roles(self.profile, progress)

    def _brief(self, state, task, progress, role, report, identifier):
        template = self.inputs[self.document["briefs"][role]["path"]].decode("utf-8")
        reports, dependencies = workflow_readiness.dependencies(state, supervision.load(self.path), self.document, progress, role, state_path=self.path)
        context = {"schema_version": 1, "task": task["task"], "scope": task["scope"],
            "authorization": task["authorization"], "base_revision": task["base_revision"],
            "allowed_paths": task["allowed_paths"], "repository": self.document["repository"],
            "head_revision": progress["head"]["commit"] if progress["head"] else None,
            "correction": progress["round"], "profile": self.profile["profile"],
            "responsibilities": schema.responsibilities(self.profile, role),
            "acceptance": self.document["acceptance"], "blocking_reports": progress["blockers"],
            "current_reports": reports,
            "source_adoption": self.document.get("adoption"),
            "publication": self.document["publication"], "effects": self.document["effects"],
            "review_evidence": str(Path(self.document["artifact_directory"]) / "reviews-{}.json".format(identifier.replace(":", "-"))) if role == "release" else None,
            "report": report}
        body = (template.rstrip() + "\n\n## Admitted current task\n\n```json\n" +
                json.dumps(context, indent=2, sort_keys=True) + "\n```\n\n" +
                "Deliver exactly one operative WORKFLOW JSON claim under the installed workflow report contract. "
                "Use the actual full current Git HEAD, the responsibilities above and this report path. "
                "Review/test outcomes also require exactly one VERDICT line. "
                "A blocked or uncertain result stays explicit; do not invent a passing check or change the admitted scope. "
                "The final chat line must match this marker exactly:\nREPORT: " + report + "\n").encode("utf-8")
        return body, dependencies

    def _make_job(self, state, task, progress, role, plan):
        root = workspace.prepare_directory(self.document["artifact_directory"])
        grant = authority.current_grant(state, task["task"])
        assert grant is not None
        header = {"schema_version": 2, "state": str(self.path), "task": task["task"],
                  "generation": progress["generation"], "contract": grant["contract"],
                  "policy_paths": self.policy}
        common_body = (self.inputs[self.document["common"]["path"]].rstrip() +
            b"\n\n## Admitted task owner\n\n```json\n" + encoded(header) + b"```\n\n" +
            b"This assignment belongs to the actual schema-2 owner above. Read POLICY_INDEX and TASK_WORKFLOW "
            b"from the absolute policy_paths. The release responsibility also reads RELEASE_SKILL in full and "
            b"runs its complete release flow through publication; the executor owns local task teardown. "
            b"Other roles read their current task inputs and role brief. No history-directory traversal or "
            b"legacy round paperwork is a normal-path prerequisite. Existing operator holds remain binding.\n")
        common, _ = workspace.write_current(root, "common.md", common_body)
        base = "workflow-" + supervision.digest([task["task"], progress["generation"], progress["round"], progress["phase"], progress["plan"]["sha256"]])[:48]
        identifier = base + ":" + role
        artifact = identifier.replace(":", "-")
        report = str(root / "report-{}.md".format(artifact))
        if Path(report).exists() or Path(report).is_symlink():
            raise UsageError("This unsent attempt already has report bytes; preserve them and reconcile rather than adopting or overwriting another result.", {})
        body, dependencies = self._brief(state, task, progress, role, report, identifier)
        brief, _ = workspace.write_current(root, "brief-{}.md".format(artifact), body, immutable=True)
        job = {"schema_version": 1, "id": identifier, "role": role,
            "agent": plan["assignments"][role], "worker_kind": plan["worker_kinds"][role],
            "tier": {key: value for key, value in plan["tiers"][role].items() if key not in cli.PLAN_ONLY_TIER_FIELDS},
            "status": "prepared", "pane_id": None, "native_session": None, "report": report,
            "brief": brief["path"], "at": self.clock(), "transport": None, "report_evidence": None, "claim": None}
        schema.validate_job(job)
        return job, common, [common, brief, progress["plan"], *dependencies]

    def _resume(self):
        with state_lock(self.path):
            state, task, progress = self._current()
            if progress["pending"] is None:
                return self._view(progress, "waiting", "pending_intent_superseded")
            intent = progress["pending"]
            operation, job = intent["operation"], intent["job"]
            if operation == "adopt":
                from . import workflow_adoption
                member = next((row for row in supervision.load(self.path)["members"]
                               if row["id"] == self.document["adoption"]["dispatch"]), None)
                if member is not None and not member["active"]:
                    workflow_adoption.dependencies(self.path, state, supervision.load(self.path), self.document)
                    progress.update(pending=None, at=self.clock())
                    self._save(state)
                    return self._view(progress, "advanced")
                if intent["effects_started"]:
                    return self._exception(state, "pending_effect_requires_reconciliation", [intent["id"]],
                                           ("reconcile_original_effect", "return_authority"))
                # The canonical native adapter verifies original source/Git
                # evidence after this lock is released. Its before-effect
                # checkpoint then rechecks captured inputs and current owner.
            if operation == "cleanup":
                try:
                    return self.release.resume_cleanup(self, state, task, progress)
                except ForemanError as exc:
                    return self._exception(state, "cleanup_owner_refused", [intent["id"], exc.message],
                                           ("reconcile_original_effect", "return_authority"))
            if operation == "apply":
                row = next((row for row in state["recovery"]["dispatches"] if row["id"] == job["id"]), None)
                if row is not None and row["status"] == "applied":
                    # Canonical readback wins over an interrupted launcher return.
                    current = dispatch.current_job(state, supervision.load(self.path), job["id"])
                    if any(current[key] != job[key] for key in ("id", "role", "agent", "report", "brief", "worker_kind")):
                        return self._exception(state, "dispatch_readback_conflict", [job["id"]])
                    progress["jobs"][job["role"]] = job["id"]
                    progress.update(pending=None, at=self.clock())
                    self._save(state)
                    return self._view(progress, "advanced")
            elif operation == "close":
                member = next((row for row in supervision.load(self.path)["members"] if row["id"] == job["id"]), None)
                if (member is not None and not member["active"] and member["resolution"] is not None
                        and job["report_evidence"] is not None
                        and all(any(row["path"] == expected["path"] and row["sha256"] == expected["sha256"]
                                    for row in member["resolution"]["evidence"]) for expected in intent["inputs"])):
                    progress.update(pending=None, at=self.clock())
                    self._save(state)
                    return self._view(progress, "advanced")
            if intent["effects_started"]:
                return self._exception(state, "pending_effect_requires_reconciliation", [intent["id"]],
                                       ("reconcile_original_effect", "return_authority"))
        def transaction():
            return self._effect_transaction(state, intent["id"], job)
        def before_effect():
            with transaction():
                self._before_effect(state)
        try:
            with self.capability.workflow_operation(state, operation, before_effect=before_effect):
                client = self.capability.authorized_client(self.client)
                adapter = self.native(self.capability, client, state, lambda: self._save(state),
                                      self.clock, sleeper=self.sleeper, owner_transaction=transaction)
                if operation == "adopt":
                    adapter.adopt(self.document)
                    return self._view(progress, "effect_recorded")
                if operation == "apply":
                    common = str(Path(self.document["artifact_directory"]) / "common.md")
                    adapter.launch(job, self.by_kind[job["worker_kind"]], {"common": common}, state["assignments"],
                                   worktree=self.document["repository"]["worktree"])
                    # Do not trust adapter return/exit. Re-enter readback next tick.
                    return self._view(progress, "effect_recorded")
                if operation == "close":
                    adapter.close(job, accepted=job["claim"]["outcome"] in ("built", "verified", "ready", "delivered"))
                    return self._view(progress, "effect_recorded")
                if operation == "measure":
                    if intent["argv"] and (len(intent["argv"]) != 2 or intent["argv"][0] != "--now"):
                        raise UsageError("Restore the measurement intent's exact owner timestamp before starting effects.", {})
                    argv = intent["argv"] or ["--now", self.clock()]
                    args = cli.build_parser().parse_args(["measure", "--config", self.document["config"]["path"],
                                                        "--state", str(self.path), *argv])
                    _result, error = cli.cmd_measure(args, client=client)
                    # The existing measurement owner saved its own fresh state.
                    with self._checkpoint_lock():
                        state, task, progress = self._current()
                        if error is not None:
                            return self._exception(state, "measurement_incomplete", [str(self.path)])
                        progress.update(pending=None, at=self.clock())
                        self._save(state)
                        return self._view(progress, "advanced")
                raise UsageError("Restore the supported original task effect before resuming progression.", {})
        except ForemanError as exc:
            # Preserve every callback checkpoint and uncertain native effect.
            with self._checkpoint_lock():
                state, task, progress = self._current()
                return self._exception(state, "native_owner_refused", [intent["id"], exc.message],
                                       ("reconcile_original_effect", "return_authority"))

    @staticmethod
    def _stamp(member):
        return supervision.digest({"assignment": supervision.expected_assignment(member),
                                   "active": member["active"], "resolution": member["resolution"]})

    def _independent(self, state, job, jobs):
        if job["role"] not in ("reviewer", "tester"):
            return
        for other in jobs:
            if other["id"] == job["id"]:
                continue
            if other["role"] not in ("developer", "reviewer", "tester"):
                continue
            if (other["agent"] == job["agent"] or other["pane_id"] == job["pane_id"]
                    or other["native_session"] is not None and other["native_session"] == job["native_session"]):
                raise UsageError("Independent verification shares an implementation or verification native identity; preserve the reports and obtain a fresh independent responsibility.", {})
        contributors = composition_contributors(state, self.capability.task)
        if job["agent"] in contributors:
            raise UsageError("This verifier contributed to the task; preserve its excluding history and dispatch independent verification.", {})

    def _record_delivery(self, state, progress, job, observed, head):
        if workspace.capture(job["report"], limit=workspace.REPORT_LIMIT)[0] != observed["evidence"]:
            raise UsageError("The delivered report changed during observation; preserve its original bytes before deciding this attempt.", {})
        if progress["head"] is not None and job["role"] != "developer" and head != progress["head"]:
            raise UsageError("The implementation changed during independent verification or release; obtain current-tip evidence before progression.", {})
        live = self.client.agent_get(job["agent"])
        if (live.get("pane_id") != job["pane_id"]
                or report_delivery.native_identity(live) != observed["native_session"]):
            raise UsageError("The native report owner changed before assessment; preserve the original session's obligation.", {})
        if job["native_session"] is None:
            supervision.refine(self.path, job["id"], job["pane_id"], observed["native_session"], self.clock())
            job["native_session"] = observed["native_session"]
        root = workspace.prepare_directory(self.document["artifact_directory"])
        delivery_name = "delivery-{}.json".format(job["id"].replace(":", "-"))
        if (root / delivery_name).exists():
            delivery, prior = workspace.capture(root / delivery_name)
            if prior != encoded(observed["wait"]):
                raise UsageError("The preserved delivery file differs from the current canonical native receipt; retain both outcomes for owner reconciliation.", {})
        else:
            delivery, _ = workspace.write_current(root, delivery_name, encoded(observed["wait"]), immutable=True)
        # This event is backed by the canonical native delivery checker, never
        # by a file's existence or the worker's claim alone.
        def observed_event(current):
            member = next(row for row in current["members"] if row["id"] == job["id"])
            detail = {"present": True, **observed["evidence"], "size": len(observed["body"].encode("utf-8"))}
            if member["observed"].get("report") != detail:
                supervision.append_event(current, self.clock(), job["id"], "report_observed", detail)
                member["observed"]["report"] = detail
        supervision.transaction(self.path, observed_event)
        if job["role"] in ("reviewer", "tester"):
            engagement.record_assessment(state, self.path, {"id": job["id"], "dispatch": job["id"],
                "report": job["report"], "delivery": delivery["path"]}, self.clock())
            engagement.require_accepted(state, job["id"], job["report"])
        # Record additive contribution evidence even when the later workflow
        # claim is malformed. A format failure cannot erase independence facts.
        claim = schema.claim(observed["body"], self.capability.task, job["role"], self.profile,
                             head["commit"], self.document["acceptance"]["criteria"])
        if job["role"] == "developer" and claim["outcome"] == "built" and not head["paths"]:
            raise UsageError("No authorized implementation change is present; reconcile the task's actual outcome instead of accepting an empty built claim.", {})
        job.update(report_evidence=observed["evidence"], claim=claim, status="delivered")
        if job["role"] in ("reviewer", "tester"):
            if claim["outcome"] != "verified":
                report_gates.record_verdict(self.path, job["report"], job["report_evidence"]["sha256"],
                                            job["id"], self.clock())
            if job["role"] == "reviewer":
                recovery.record_report(state["recovery"], {"dispatch": progress["jobs"]["developer"],
                    "head_revision": head["commit"], "verdict": "approved" if claim["outcome"] == "verified" else "blocking",
                    "review_mode": "full", "reviewer": job["agent"], "report": job["report"],
                    "changed_paths": head["paths"]}, self.clock())
        if job["role"] == "developer" and claim["outcome"] == "built":
            progress["head"] = head
        self._intent(progress, "close", job=job, inputs=[observed["evidence"], delivery], dispatches=[job["id"]])
        progress["at"] = self.clock()
        self._save(state)

    def _route(self, state, task, progress, jobs, head):
        claims, receipts = {}, {}
        supervised = supervision.load(self.path)
        for job in jobs:
            if job["role"] not in self._roles(progress):
                continue
            receipt, body = workspace.capture(job["report"], limit=workspace.REPORT_LIMIT)
            member = next(row for row in supervised["members"] if row["id"] == job["id"])
            if (job["status"] != "closed" or member["resolution"] is None or not any(
                    item["path"] == receipt["path"] and item["sha256"] == receipt["sha256"]
                    for item in member["resolution"]["evidence"])):
                raise UsageError("Restore the closed assignment's byte-bound report before routing its outcome or correction inputs.", {})
            claims[job["role"]] = schema.claim(body.decode(), task["task"], job["role"], self.profile,
                head["commit"], self.document["acceptance"]["criteria"])
            receipts[job["role"]] = receipt
        if any(claim["outcome"] == "blocked" for claim in claims.values()):
            return self._exception(state, "worker_blocked", [job["report"] for job in jobs if job["role"] in claims])
        if head != progress["head"]:
            return self._exception(state, "verified_head_changed", [self.document["repository"]["worktree"]])
        if progress["phase"] == "implement":
            progress.update(phase="verify", plan=None, at=self.clock())
        elif progress["phase"] in ("verify", "release"):
            routes = [schema.workflow_profiles.correction_route({"schema_version": 1,
                "authority_current": True, "evidence_current": True, **claim["correction"]}, claim["findings"])
                for claim in claims.values() if claim["correction"] is not None]
            if any(route["action"] == "decision_packet" for route in routes):
                return self._exception(state, "verification_decision_required", [job["report"] for job in jobs if job["role"] in claims],
                                       ("resolve_task_decision", "return_authority"))
            if any(claim["outcome"] == "blocking" for claim in claims.values()):
                budget = recovery.task_statuses(state["recovery"], state["assignments"])[task["task"]]
                if budget["remaining_fixes"] <= 0 or budget["status"] == "diagnosed_stop":
                    return self._exception(state, "correction_allowance_exhausted", [job["report"] for job in jobs if job["role"] in claims],
                                           ("diagnose_task", "return_authority"))
                blockers = [receipts[job["role"]]
                            for job in jobs if job["role"] in claims and claims[job["role"]]["outcome"] == "blocking"]
                progress.update(phase="implement", round=progress["round"] + 1, jobs={}, plan=None,
                                blockers=blockers, at=self.clock())
            elif progress["phase"] == "verify":
                # Re-checks clear only the same responsibility's actual verdict
                # gates, through the existing evidence owner.
                for job in jobs:
                    if job["role"] not in claims:
                        continue
                    gated = report_gates.open_verdict_gates(self.path, task["task"], state["recovery"]["dispatches"])
                    for gate in gated:
                        gated_dispatch = next(row for row in state["recovery"]["dispatches"] if row["id"] == gate["dispatch"])
                        if gated_dispatch["role"] == job["role"]:
                            report_gates.resolve(self.path, gate["report"], "clear", "Independent current-tip workflow re-check.",
                                self.clock(), report_gates.ledger_view(self.path, None, gate["report"]), evidence=job["report"])
                report_gates.require_no_verdict_gate(self.path, task["task"], state["recovery"]["dispatches"])
                progress.update(phase="release", plan=None, blockers=[], at=self.clock())
            elif claims["release"]["outcome"] == "ready":
                # Preparation is settled, but its immutable report cannot
                # become delivery. Continue with a fresh release assignment.
                progress["jobs"].pop("release")
                progress.update(plan=None, at=self.clock())
            else:
                workflow_upgrade.require_compatible_services(state, self.path)
                progress.update(phase="merge", plan=None, at=self.clock(), delivery={
                    "schema_version": 2, "pull_request": claims["release"]["pull_request"],
                    "merge_commit": None, "publisher_run": None, "tag": None, "tessl_version": None,
                    "tessl_confirmed": False, "acr_confirmed": False, "accepted": False,
                    "base_checkout": str(workspace.base_checkout(self.document)), "cleanup_step": 0,
                    "artifacts": workflow_artifacts.initial()})
        self._save(state)
        return self._view(progress, "advanced")

    def tick(self):
        authority.verify_foreman(self.path, self.client, native_only=True)
        try:
            return self._tick()
        except TransactionBusy:
            return load_set._bounded({"schema_version": 2, "wake_agent": False, "data": {
                "task": self.capability.task, "generation": self.capability.generation,
                "status": "waiting", "reason": "owner_transaction_busy",
                "authorizes_acceptance": False, "authorizes_release": False}})
        except ForemanError as exc:
            with state_lock(self.path):
                state, _task, _progress = self._current()
                return self._exception(state, "workflow_owner_refused", [exc.message],
                                       ("inspect_current_evidence", "return_authority"))

    def _tick(self):
        if (getattr(getattr(self.client, "context", None), "mode", None) != "native"
                or not os.environ.get("HERDR_ENV")):
            raise UsageError("Run the task owner in its actual connected native environment; saved state never supplies HERDR_ENV.", {})
        progression_runtime.pulse(self.capability, self.clock())
        stamp = ""
        with state_lock(self.path):
            state, task, progress = self._current()
            if progress["phase"] in ("exception", "complete"):
                return self._view(progress, progress["phase"])
            pending = progress["pending"] is not None
            if not pending and self.profile["specialist_domains"]:
                return self._exception(state, "specialist_evidence_required", self.profile["specialist_domains"],
                                       ("admit_specialist_evidence", "return_authority"))
            roles = self._roles(progress)
            if not roles:
                stamp = supervision.digest(progress)
                release_observation = True
            else:
                release_observation = False
        if pending:
            return self._resume()
        if release_observation:
            # Review/publication helpers own their waits. Their potentially
            # long reads cannot retain another task's primary transaction.
            try:
                observed = self.release.observe(self, state, task, progress)
            except ForemanError as exc:
                with state_lock(self.path):
                    state, task, progress = self._current()
                    return self._exception(state, "release_evidence_unconfirmed", [exc.message],
                                           ("reconcile_release_evidence", "return_authority"))
            with state_lock(self.path):
                state, task, progress = self._current()
                if supervision.digest(progress) != stamp:
                    return self._view(progress, "waiting", "release_observation_superseded")
                try:
                    return self.release.commit(self, state, task, progress, observed)
                except ForemanError as exc:
                    return self._exception(state, "release_evidence_unconfirmed", [exc.message],
                                           ("reconcile_release_evidence", "return_authority"))
        with state_lock(self.path):
            state, task, progress = self._current()
            roles = self._roles(progress)
            missing = [role for role in roles if role not in progress["jobs"]]
            preparation_stamp = supervision.digest(progress)
            supervised = supervision.load(self.path)
        if missing:
            # Git, planning and artifact rendering are task-local preparation.
            # Another task can prepare concurrently; the task's kernel lease
            # provides the single writer, and only its reservation is global.
            decision = workflow_readiness.preparation(state, supervised, self.path,
                self.document, progress, roles, self.agents, self.backpressure, self.clock(), planner=self.planner)
            capacity = decision["capacity"]
            if decision["state"] == "capacity_wait":
                assert capacity is not None
                return self._view(progress, "waiting", capacity["reason"], capacity=capacity)
            if decision["state"] == "measurement_required":
                with state_lock(self.path):
                    state, task, progress = self._current()
                    if supervision.digest(progress) != preparation_stamp:
                        return self._view(progress, "waiting", "preparation_superseded")
                    self._intent(progress, "measure", inputs=[self.document["config"]])
                    self._save(state)
                    return self._view(progress, "prepared", capacity=capacity)
            if decision["state"] == "plan_ready":
                prepared_plan, _ = workspace.write_current(self.document["artifact_directory"], "plan.json", encoded(decision["plan"]))
                with state_lock(self.path):
                    state, task, progress = self._current()
                    if supervision.digest(progress) != preparation_stamp:
                        return self._view(progress, "waiting", "preparation_superseded")
                    progress["plan"] = prepared_plan
                    self._save(state)
                    preparation_stamp = supervision.digest(progress)
            try:
                plan = workflow_readiness.read_plan(progress["plan"], missing, self.by_kind)
            except ForemanError as exc:
                with state_lock(self.path):
                    state, task, progress = self._current()
                    return self._exception(state, "current_plan_changed", [exc.message])
            job, common, inputs = self._make_job(state, task, progress, missing[0], plan)
            prepared_plan = progress["plan"]
            with state_lock(self.path):
                state, task, progress = self._current()
                if supervision.digest(progress) != preparation_stamp:
                    return self._view(progress, "waiting", "preparation_superseded")
                workflow_planning.require_headroom(state, plan["worker_kinds"][missing[0]], self.clock())
                if self.backpressure is not None:
                    from .workflow_capacity import evaluate
                    current_capacity = evaluate(state, supervision.load(self.path), self.capability.task,
                        self.document["repository"]["github"], self.backpressure, self.clock(),
                        family=lambda name: workflow_readiness.capacity_family(state, name))
                    if not current_capacity["allowed"]:
                        return self._view(progress, "waiting", current_capacity["reason"], capacity=current_capacity)
                progress["plan"] = prepared_plan
                dispatch.reserve(state, task["task"], progress["generation"], progress["round"], job,
                    self.by_kind[job["worker_kind"]], common["path"], self.clock())
                self._intent(progress, "apply", job=job, inputs=inputs, dispatches=[job["id"]])
                self._save(state)
                return self._view(progress, "prepared", capacity=capacity)
        with state_lock(self.path):
            state, task, progress = self._current()
            roles = self._roles(progress)
            supervised = supervision.load(self.path)
            jobs = [dispatch.current_job(state, supervised, identifier) for identifier in progress["jobs"].values()]
            for job in jobs:
                self._independent(state, job, jobs)
            active = [job for job in jobs if job["role"] in roles and job["status"] != "closed"]
            if not active:
                # All current source checks happen again before the transition.
                head = workspace.inspect(self.document, task)
                try:
                    return self._route(state, task, progress, jobs, head)
                except ForemanError as exc:
                    return self._exception(state, "evidence_owner_refused", [exc.message])
            expected_progress = supervision.digest(progress)
            stamps = {job["id"]: self._stamp(next(row for row in supervised["members"] if row["id"] == job["id"])) for job in active}
        # Never hold the primary task lock while waiting for native completion.
        adapter = self.native(self.capability, self.client, state, lambda: None, self.clock, sleeper=self.sleeper)
        for job in active:
            observed = adapter.observe(job, self.by_kind[job["worker_kind"]], task,
                                       self.document["repository"]["worktree"])
            if observed["state"] == "waiting":
                continue
            with state_lock(self.path):
                state, task, progress = self._current()
                current = next(row for row in supervision.load(self.path)["members"] if row["id"] == job["id"])
                if supervision.digest(progress) != expected_progress or self._stamp(current) != stamps[job["id"]]:
                    return self._view(progress, "waiting", "observation_superseded")
                if observed["state"] != "delivered":
                    return self._exception(state, "native_delivery_exception", [job["report"], observed["reason"]],
                                           ("reconcile_original_effect", "return_authority"))
                try:
                    head = workspace.inspect(self.document, task, built=job["role"] != "developer")
                    self._record_delivery(state, progress, job, observed, head)
                except ForemanError as exc:
                    return self._exception(state, "report_contract_refused", [job["report"], exc.message])
                return self._view(progress, "prepared")
        return self._view(progress, "waiting", "native_work_in_progress")


def composition_contributors(state, task):
    """Actual native contributors, including declared contributions on failed reports."""
    contributors = {row["agent"] for row in state["recovery"]["dispatches"]
                    if row["task"] == task and row["role"] == "developer" and row["status"] != "not_sent"}
    contributors.update(row["agent"] for row in state["specialist_assessments"]
                        if row["task"] == task and row["contribution"] in engagement.EXCLUDING_CONTRIBUTIONS)
    return contributors