Index

async def run_step(
    self,
    step_id: str,
    step_type: str,
    fn: Callable[[], Awaitable[dict[str, Any]] | dict[str, Any]],
    *,
    goal_id: str = "",
    task_id: str = "",
    thread_id: str = "",
    actor: str = "lead",
    causal_parents: list[str] | None = None,
) -> StepResult:
    """Execute (or replay) one durable step (AC-4-3, AC-4-4, AC-4-5).

    If a result event for ``step_id`` already exists in the log, returns it
    without re-running ``fn`` (idempotent replay, AC-4-4). Otherwise awaits
    ``fn`` (which may be sync or async — an async callable is awaited, a sync
    one is called directly; neither blocks the loop via ``asyncio.run``,
    AC-4-5), emits a ``step.result`` :class:`~strands_teams.models.TeamEvent`
    recording the transition and result to the log (AC-4-3), and returns the
    fresh :class:`StepResult`.

    Args:
        step_id: Stable idempotency key for the step within the goal run.
        step_type: The lead-loop transition (e.g. ``decompose``).
        fn: A zero-arg callable producing the step's result payload dict; may be
            sync or async.
        goal_id: The goal this step belongs to.
        task_id: The task this step belongs to.
        thread_id: The thread this step belongs to.
        actor: Identifier of the step's executor (defaults to ``"lead"``).
        causal_parents: ``event_id``s this step's event causally follows.

    Returns:
        A :class:`StepResult` — ``replayed=True`` when returned from a recorded
        event, else the freshly executed result.
    """
    recorded = self._recorded_result(step_id)
    if recorded is not None:
        logger.debug(f"AsyncioDurableEngine: step {step_id!r} already recorded — replaying without re-running")
        return recorded

    outcome = fn()
    awaited = await outcome if inspect.isawaitable(outcome) else outcome
    payload = cast("dict[str, Any]", awaited)
    result_payload: dict[str, Any] = dict(payload) if payload else {}

    event = TeamEvent(
        event_type=STEP_RESULT_EVENT_PREFIX,
        actor=actor,
        goal_id=goal_id,
        task_id=task_id,
        thread_id=thread_id,
        causal_parents=list(causal_parents or []),
        epistemic_status="observed",
        payload={"step_id": step_id, "step_type": step_type, "result": result_payload},
    )
    self._event_log.append(event)
    return StepResult(step_id=step_id, step_type=step_type, payload=result_payload, replayed=False)

async def run_step(
    self,
    step_id: str,
    step_type: str,
    fn: Callable[[], Awaitable[dict[str, Any]] | dict[str, Any]],
    *,
    goal_id: str = "",
    task_id: str = "",
    thread_id: str = "",
    actor: str = "lead",
    causal_parents: list[str] | None = None,
) -> StepResult:
    """Execute one step and record its transition result; return the outcome."""
    ...

def append(self, event: TeamEvent) -> None:
    """Append exactly one JSON line for ``event``; never raise (AC-2-2, AC-2-4).

    Serializes ``event`` to a single line via
    :meth:`~pydantic.BaseModel.model_dump_json` and appends it, leaving every
    previously written line byte-for-byte unchanged. An ``OSError`` (e.g. a
    read-only filesystem) is logged and swallowed so a log-write failure never
    kills the coordinator or the lead loop (AC-2-4), matching the never-raises
    contract of the pre-split ``_write_debug_entry`` (``teams.py``).
    """
    try:
        self._path.parent.mkdir(parents=True, exist_ok=True)
        line = event.model_dump_json()
        with self._path.open("a", encoding="utf-8") as f:
            f.write(line + "\n")
    except OSError as e:  # AC-2-4: log and continue, never raise.
        logger.debug(f"EventLog.append failed for {self._path}: {e}")

def read(self) -> list[TeamEvent]:
    """Parse and return every valid event, skipping malformed lines.

    A malformed line is skipped; use :func:`replay` when you need the
    malformed-line warnings surfaced structurally (AC-3-5).
    """
    events, _ = parse_lines(self._read_raw_lines())
    return events

def read_with_warnings(self) -> tuple[list[TeamEvent], list[str]]:
    """Return ``(events, warnings)`` — one warning per skipped malformed line."""
    return parse_lines(self._read_raw_lines())

def to_canonical_json(self) -> str:
    """Return a byte-stable JSON serialization for cross-replay equality (AC-3-2)."""
    return json.dumps(self.model_dump(mode="json"), sort_keys=True, ensure_ascii=False)

def teams_dir(self) -> Path:
    """Directory holding every team's output subdirectory."""
    return self.output_root / self.scratch_subdir / self.teams_subdir

def team_dir(self, team_id: str) -> Path:
    """Directory for a single team's output."""
    return self.teams_dir() / team_id

def findings_path(self, team_id: str) -> Path:
    """Path to a team's ``findings.md``."""
    return self.team_dir(team_id) / self.findings_name

def findings_json_path(self, team_id: str) -> Path:
    """Path to a team's typed ``findings.json`` sidecar.

    Resolves under the same per-team directory as :meth:`findings_path`, so
    the human markdown and the machine-readable typed findings live side by
    side (AC-2-2).
    """
    return self.team_dir(team_id) / self.findings_json_name

def contract_path(self, team_id: str) -> Path:
    """Path to a team's task-contract sidecar (``contract.json``).

    Resolves under the same per-team directory as :meth:`findings_path`, so a
    team's granted authority, budget, acceptance criteria, and input
    revisions are persisted next to its findings and metadata (AC-2-6). The
    :class:`~strands_teams.hooks.ArtifactQualityHook` reads every team's
    ``acceptance_criteria`` from these files to evaluate criteria coverage.
    """
    return self.team_dir(team_id) / self.contract_name

def contract_glob(self) -> str:
    """Glob (relative to :meth:`teams_dir`) matching every per-team contract file."""
    return f"*/{self.contract_name}"

def metadata_path(self, team_id: str) -> Path:
    """Path to a team's ``metadata.json``."""
    return self.team_dir(team_id) / self.metadata_name

def debug_log_path(self) -> Path:
    """Path to the append-only team-dispatch debug JSONL."""
    return self.teams_dir() / self.debug_log_name

def event_log_path(self) -> Path:
    """Path to the typed append-only event-plane JSONL (Phase 5 / spec 006).

    Co-located with :meth:`debug_log_path` under the teams dir so the typed
    event plane supersedes the untyped ``team_debug.jsonl`` without moving the
    run's scratch layout (AC-2-1). One JSON-encoded
    :class:`~strands_teams.models.TeamEvent` per line.
    """
    return self.teams_dir() / self.event_log_name

def artifacts_dir(self) -> Path:
    """Directory holding per-area artifact files."""
    return self.output_root / self.artifacts_subdir

def artifact_path(self, area: str) -> Path:
    """Path to a per-area artifact file (e.g., ``bundles/BUNDLE.auth.md``)."""
    return self.artifacts_dir() / f"{self.artifact_prefix}.{area}{self.artifact_suffix}"

def artifact_glob(self) -> str:
    """Glob pattern matching every per-area artifact file."""
    return f"{self.artifact_prefix}.*{self.artifact_suffix}"

def area_from_artifact(self, path: Path) -> str:
    """Extract the area identifier from an artifact file path.

    Inverse of :meth:`artifact_path` — strips the prefix/suffix so the
    quality hook can name a weak artifact by its area.
    """
    return path.name.removeprefix(f"{self.artifact_prefix}.").removesuffix(self.artifact_suffix)

def executive_summary_path(self) -> Path:
    """Path to the top-level executive summary file."""
    return self.output_root / self.executive_summary_name

def planning_sentinel_path(self) -> Path:
    """Path to the planning-context sentinel produced by the pre-run step."""
    return self.output_root / self.planning_sentinel_name

def revisions_dir(self) -> Path:
    """Directory holding the content-addressed immutable revision store (AC-1-3)."""
    return self.output_root / self.revisions_subdir

def revision_path(self, logical_name: str, revision_id: str) -> Path:
    """Resolve the on-disk path for one ``(logical_name, revision_id)`` revision (AC-1-3).

    Every revision of a logical artifact is grouped under a per-artifact
    subdirectory of :meth:`revisions_dir`, keyed by ``revision_id`` (which is
    content-addressed). This is the ONLY place a revision path is derived, so
    dispatch/read/write/promotion code never duplicates the literal (AC-1-3).
    """
    return self.revisions_dir() / _slug(logical_name) / f"{_slug(revision_id)}.json"

def alias_path(self) -> Path:
    """Path to the mutable ``logical_name -> current revision_id`` alias map (AC-2-1).

    Resolved directly under ``output_root`` in the same manner as
    :meth:`planning_sentinel_path` — a single durable JSON file that is the
    only mutable pointer into the content-addressed revision store.
    """
    return self.output_root / self.alias_name

def persist_revision(self, revision: ArtifactRevision) -> ArtifactRevision:
    """Persist a revision to its content-addressed path and return it (AC-1-4, AC-1-5).

    Writes the revision keyed by ``revision_id`` under :meth:`revision_path` so
    it is addressable by later reads (AC-1-4). Idempotent for an identical
    re-write (same ``revision_id`` and ``content_hash``). If the same
    ``revision_id`` was already persisted with a DIFFERENT ``content_hash``,
    the write is rejected with :class:`RevisionConflictError` and the stored
    revision is left unchanged (AC-1-5) — the store is append-only and
    content-addressed.

    Returns:
        The stored :class:`ArtifactRevision` (the argument on a fresh write, or
        the already-stored record on an idempotent re-write).

    Raises:
        RevisionConflictError: If ``revision.revision_id`` is already stored
            with content differing from ``revision.content_hash`` (AC-1-5).
    """
    path = self.revision_path(revision.logical_name, revision.revision_id)
    if path.exists():
        stored = ArtifactRevision.model_validate_json(path.read_text(encoding="utf-8"))
        if stored.content_hash != revision.content_hash:
            raise RevisionConflictError(revision.revision_id, stored.content_hash, revision.content_hash)
        return stored
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(revision.model_dump_json(indent=2), encoding="utf-8")
    return revision

def load_revision(self, logical_name: str, revision_id: str) -> ArtifactRevision | None:
    """Load a persisted revision, or ``None`` when it is not in the store."""
    path = self.revision_path(logical_name, revision_id)
    if not path.exists():
        return None
    return ArtifactRevision.model_validate_json(path.read_text(encoding="utf-8"))

def current_revision_id(self, logical_name: str) -> str:
    """Return the ``revision_id`` the alias currently points ``logical_name`` at.

    Returns the :data:`~strands_teams.models.NO_CURRENT_REVISION` sentinel when
    ``logical_name`` has no current revision (AC-2-4).
    """
    with self._locked_aliases() as aliases:
        return aliases.get(logical_name, NO_CURRENT_REVISION)

def promote_revision(
    self,
    logical_name: str,
    expected_current_revision: str,
    proposed_revision: ArtifactRevision,
) -> ArtifactRevision:
    """Durably compare-and-set the alias for ``logical_name`` (AC-2-2..AC-2-5).

    The compare-and-set is serialized by an OS advisory lock on the alias file,
    so two concurrent promotions of the same ``logical_name`` with the same
    ``expected_current_revision`` cannot both win: whichever acquires the lock
    first flips the alias, and the other observes the new current and is
    rejected with :class:`StaleArtifactError` (AC-2-5). Because the CAS is a
    durable operation against this on-disk store — not a mutation of the
    by-reference ``invocation_state`` dict — it is correct even when the callers
    share that dict read-only (research-coding-runtimes.yaml lines 22, 30).

    On a matching CAS the proposed revision is persisted (content-addressed)
    with the prior current revision recorded in its ``supersedes`` lineage, and
    the alias is atomically flipped to it (AC-2-2). A first promotion passes the
    :data:`~strands_teams.models.NO_CURRENT_REVISION` sentinel as
    ``expected_current_revision`` and sets the alias as the first revision
    (AC-2-4).

    Args:
        logical_name: The logical artifact to promote.
        expected_current_revision: The revision the caller believes is current
            (the sentinel for a first promotion).
        proposed_revision: The revision to make current. Its ``supersedes`` is
            augmented with the prior current revision before it is persisted.

    Returns:
        The persisted, now-current :class:`ArtifactRevision` (with the prior
        revision recorded in ``supersedes``).

    Raises:
        StaleArtifactError: If ``expected_current_revision`` does not match the
            alias's stored current revision — the alias is left unchanged
            (AC-2-3).
        RevisionConflictError: If persisting the proposed revision conflicts
            with an already-stored revision of the same id (AC-1-5).
    """
    with self._locked_aliases() as aliases:
        actual = aliases.get(logical_name, NO_CURRENT_REVISION)
        if actual != expected_current_revision:
            raise StaleArtifactError(logical_name, expected_current_revision, actual)

        superseded = [] if actual == NO_CURRENT_REVISION else [actual]
        merged_supersedes = list(dict.fromkeys([*proposed_revision.supersedes, *superseded]))
        to_store = proposed_revision.model_copy(update={"supersedes": merged_supersedes})

        stored = self.persist_revision(to_store)
        aliases[logical_name] = stored.revision_id
        self._write_aliases(aliases)
        return stored

def approvals_dir(self) -> Path:
    """Directory holding the per-``token_id`` :class:`ApprovalToken` store (AC-3-7)."""
    return self.output_root / self.scratch_subdir / self.approvals_subdir

def approval_path(self, token_id: str) -> Path:
    """Resolve the on-disk path for one approval token, keyed by ``token_id``."""
    return self.approvals_dir() / f"{_slug(token_id)}.json"

def persist_approval(self, token: ApprovalToken) -> ApprovalToken:
    """Persist an :class:`ApprovalToken` to the approvals store and return it (AC-1-1)."""
    path = self.approval_path(token.token_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(token.model_dump_json(indent=2), encoding="utf-8")
    return token

def load_approval(self, token_id: str) -> ApprovalToken | None:
    """Load one persisted approval token, or ``None`` when absent/unreadable."""
    path = self.approval_path(token_id)
    if not path.exists():
        return None
    try:
        return ApprovalToken.model_validate_json(path.read_text(encoding="utf-8"))
    except (ValueError, OSError):
        return None

def load_approvals(self) -> list[ApprovalToken]:
    """Load every persisted approval token (skipping any that fail to parse — AC-3-7)."""
    approvals_dir = self.approvals_dir()
    if not approvals_dir.exists():
        return []
    tokens: list[ApprovalToken] = []
    for path in sorted(approvals_dir.glob("*.json")):
        try:
            tokens.append(ApprovalToken.model_validate_json(path.read_text(encoding="utf-8")))
        except (ValueError, OSError):
            continue
    return tokens

def decision_log_path(self) -> Path:
    """Path to the append-only decision log — one :class:`Decision` per line (AC-2-4)."""
    return self.output_root / self.scratch_subdir / self.decision_log_name

def append_decision(self, decision: Decision) -> None:
    """Append one :class:`Decision` to the decision log (AC-2-4).

    Raises on I/O failure rather than swallowing it, so a transition can never
    proceed with its decision unrecorded (AC-2-5). The caller decides how to
    react; unlike the debug log, this path is auth-critical.

    Raises:
        OSError: If the decision line cannot be written.
    """
    path = self.decision_log_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as f:
        f.write(decision.model_dump_json() + "\n")

def human_gates_dir(self) -> Path:
    """Directory holding pending :class:`HumanDecision` gate sentinels (AC-4-3, AC-4-5)."""
    return self.output_root / self.scratch_subdir / self.human_gates_subdir

def human_gate_path(self, decision_id: str) -> Path:
    """Resolve the on-disk path for one human-gate sentinel, keyed by ``decision_id``."""
    return self.human_gates_dir() / f"{_slug(decision_id)}.json"

def persist_human_gate(self, gate: HumanDecision) -> HumanDecision:
    """Persist a :class:`HumanDecision` gate to disk and return it (AC-4-5).

    A pending gate persisted here is what lets a swarm node return control
    rather than block past its ``node_timeout`` (AC-4-5): the gate survives on
    disk and the runner resumes from it later.
    """
    path = self.human_gate_path(gate.decision_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(gate.model_dump_json(indent=2), encoding="utf-8")
    return gate

def load_human_gate(self, decision_id: str) -> HumanDecision | None:
    """Load one persisted human-gate sentinel, or ``None`` when absent/unreadable."""
    path = self.human_gate_path(decision_id)
    if not path.exists():
        return None
    try:
        return HumanDecision.model_validate_json(path.read_text(encoding="utf-8"))
    except (ValueError, OSError):
        return None

def load_pending_human_gates(self) -> list[HumanDecision]:
    """Load every human gate still ``pending`` (the runner suspends on these — AC-4-3)."""
    gates_dir = self.human_gates_dir()
    if not gates_dir.exists():
        return []
    pending: list[HumanDecision] = []
    for path in sorted(gates_dir.glob("*.json")):
        try:
            gate = HumanDecision.model_validate_json(path.read_text(encoding="utf-8"))
        except (ValueError, OSError):
            continue
        if gate.status == "pending":
            pending.append(gate)
    return pending

@classmethod
def mint(
    cls,
    *,
    approved_transition_kind: str,
    artifact_revision_id: str,
    revision_content_hash: str,
    approver: str,
    signing_key: str,
    valid_until: datetime | None = None,
    valid_for: timedelta | None = None,
    invalidation_rules: list[str] | None = None,
    token_id: str | None = None,
    granted_at: datetime | None = None,
) -> ApprovalToken:
    """Mint a signed :class:`ApprovalToken` over its canonical payload (AC-1-3).

    Computes the HMAC ``signature`` with ``signing_key`` so the token verifies
    against the same key later (AC-1-4). Supply the expiry either as an absolute
    ``valid_until`` or a relative ``valid_for`` (added to ``granted_at``); when
    neither is given the token is valid for 24 hours (AC-1-5).

    Args:
        approved_transition_kind: The transition kind approved.
        artifact_revision_id: The approved revision's id.
        revision_content_hash: The approved revision's content hash (AC-1-2).
        approver: The approving principal.
        signing_key: The configured signing key (from ``TeamsSettings``).
        valid_until: Absolute UTC expiry (wins over ``valid_for``).
        valid_for: Relative validity window from ``granted_at``.
        invalidation_rules: Facet names whose change sunsets the approval.
        token_id: Explicit id; defaults to a fresh UUID hex.
        granted_at: Explicit grant time; defaults to now.

    Returns:
        A signed, frozen :class:`ApprovalToken`.
    """
    resolved_id = token_id or uuid4().hex
    resolved_granted = granted_at or _utcnow()
    if valid_until is None:
        valid_until = resolved_granted + (valid_for if valid_for is not None else timedelta(hours=24))
    payload = approval_signing_payload(
        token_id=resolved_id,
        approved_transition_kind=approved_transition_kind,
        artifact_revision_id=artifact_revision_id,
        revision_content_hash=revision_content_hash,
        approver=approver,
        valid_until=valid_until,
    )
    return cls(
        token_id=resolved_id,
        approved_transition_kind=approved_transition_kind,
        artifact_revision_id=artifact_revision_id,
        revision_content_hash=revision_content_hash,
        approver=approver,
        granted_at=resolved_granted,
        valid_until=valid_until,
        signature=_sign(payload, signing_key),
        invalidation_rules=list(invalidation_rules or []),
    )

def signing_payload(self) -> str:
    """Return this token's canonical signing payload (AC-1-3)."""
    return approval_signing_payload(
        token_id=self.token_id,
        approved_transition_kind=self.approved_transition_kind,
        artifact_revision_id=self.artifact_revision_id,
        revision_content_hash=self.revision_content_hash,
        approver=self.approver,
        valid_until=self.valid_until,
    )

def signature_valid(self, signing_key: str) -> bool:
    """Return whether ``signature`` verifies against ``signing_key`` (AC-1-4).

    Uses :func:`hmac.compare_digest` for a constant-time comparison so a
    signature check cannot be timing-attacked. A token whose signature does not
    verify is treated as absent by the ``PolicyEngine``.
    """
    expected = _sign(self.signing_payload(), signing_key)
    return hmac.compare_digest(expected, self.signature)

def is_expired(self, now: datetime | None = None) -> bool:
    """Return whether the token is at or past its ``valid_until`` (AC-1-5)."""
    return (now or _utcnow()) >= self.valid_until