Workflows API Reference

cs_copilot.workflows

Import-safe workflow policy helpers and reusable workflow contracts.

WorkflowRegistry

Discover and serve local workflow contracts.

Source code in src/cs_copilot/workflows/registry.py

class WorkflowRegistry:
    """Discover and serve local workflow contracts."""

    def __init__(self, root: str | os.PathLike[str] | None = None) -> None:
        self.root = Path(root).resolve() if root is not None else discover_workflow_root()
        self._workflows: dict[str, WorkflowSpec] | None = None

    def list_workflows(self) -> list[WorkflowSpec]:
        return list(self._load().values())

    def get_workflow(self, slug: str) -> WorkflowSpec:
        normalized = _normalize_slug(slug)
        try:
            return self._load()[normalized]
        except KeyError as exc:
            available = ", ".join(sorted(self._load()))
            raise KeyError(f"Unknown workflow '{slug}'. Available workflows: {available}") from exc

    def search_workflows(self, query: str, *, limit: int = 10) -> list[WorkflowSpec]:
        terms = _tokenize(query)
        query_lc = (query or "").lower()
        scored: list[tuple[int, WorkflowSpec]] = []
        for index, spec in enumerate(self.list_workflows()):
            haystack = _search_text(spec)
            if not terms:
                score = max(1, 100 - index)
            else:
                score = sum(_term_score(term, haystack, spec) for term in terms)
                score += _keyword_score(query_lc, spec.keywords)
            if score > 0:
                scored.append((score, spec))
        scored.sort(key=lambda item: (-item[0], item[1].slug))
        return [spec for _score, spec in scored[:limit]]

    def _load(self) -> dict[str, WorkflowSpec]:
        if self._workflows is not None:
            return self._workflows
        if not self.root.exists():
            raise FileNotFoundError(f"Workflow catalog root does not exist: {self.root}")
        workflows: dict[str, WorkflowSpec] = {}
        for path in sorted(item for item in self.root.iterdir() if item.is_dir()):
            spec = _load_workflow(path)
            if spec.slug in workflows:
                raise ValueError(f"Duplicate workflow slug: {spec.slug}")
            workflows[spec.slug] = spec
        _validate_catalog_dependencies(workflows)
        self._workflows = workflows
        return workflows

WorkflowSpec dataclass

Metadata and content for one reusable workflow contract.

Source code in src/cs_copilot/workflows/registry.py

@dataclass(frozen=True)
class WorkflowSpec:
    """Metadata and content for one reusable workflow contract."""

    slug: str
    title: str
    summary: str
    version: str
    status: str
    tags: tuple[str, ...]
    keywords: tuple[str, ...]
    depends_on: tuple[str, ...]
    profiles: tuple[str, ...]
    permissions: tuple[str, ...]
    input_artifacts: tuple[ArtifactContract, ...]
    output_artifacts: tuple[ArtifactContract, ...]
    tasks: tuple[TaskContract, ...]
    preflight_tools: tuple[str, ...]
    required_tools: tuple[str, ...]
    optional_tools: tuple[str, ...]
    expected_artifacts: tuple[str, ...]
    recommended_prompt: str | None
    workflow_md: str
    path: Path

    def as_dict(self, *, include_content: bool = False) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "slug": self.slug,
            "title": self.title,
            "summary": self.summary,
            "version": self.version,
            "status": self.status,
            "tags": list(self.tags),
            "keywords": list(self.keywords),
            "depends_on": list(self.depends_on),
            "profiles": list(self.profiles),
            "permissions": list(self.permissions),
            "input_artifacts": [contract.as_dict() for contract in self.input_artifacts],
            "output_artifacts": [contract.as_dict() for contract in self.output_artifacts],
            "tasks": [task.as_dict() for task in self.tasks],
            "preflight_tools": list(self.preflight_tools),
            "required_tools": list(self.required_tools),
            "optional_tools": list(self.optional_tools),
            "expected_artifacts": list(self.expected_artifacts),
            "recommended_prompt": self.recommended_prompt,
        }
        if include_content:
            payload["workflow_md"] = self.workflow_md
        return payload

ArtifactIntegrityError

Bases: WorkflowRuntimeError

Raised when an artifact is missing or its checksum does not match.

Source code in src/cs_copilot/workflows/runtime.py

class ArtifactIntegrityError(WorkflowRuntimeError):
    """Raised when an artifact is missing or its checksum does not match."""

ArtifactRecord dataclass

Immutable identity and provenance for one workflow-scoped artifact.

Source code in src/cs_copilot/workflows/runtime.py

@dataclass(frozen=True)
class ArtifactRecord:
    """Immutable identity and provenance for one workflow-scoped artifact."""

    artifact_id: str
    run_id: str
    artifact_type: str
    mime_type: str
    relative_path: str
    sha256: str
    size_bytes: int
    producer_task_id: str | None = None
    producer_tool: str | None = None
    provenance: Mapping[str, Any] = field(default_factory=dict)
    trust: ArtifactTrust = ArtifactTrust.INTERNAL
    created_at: str = field(default_factory=lambda: _utc_now())

    def __post_init__(self) -> None:
        object.__setattr__(
            self, "artifact_id", validate_identifier(self.artifact_id, field="artifact_id")
        )
        object.__setattr__(self, "run_id", validate_identifier(self.run_id, field="run_id"))
        object.__setattr__(
            self,
            "relative_path",
            normalize_run_relative_path(self.run_id, self.relative_path),
        )
        object.__setattr__(self, "trust", ArtifactTrust(self.trust))
        if not self.artifact_type.strip() or not self.mime_type.strip():
            raise ValueError("artifact_type and mime_type are required")
        if self.size_bytes < 0:
            raise ValueError("artifact size cannot be negative")
        if len(self.sha256) != 64 or any(char not in "0123456789abcdef" for char in self.sha256):
            raise ValueError("artifact sha256 must be a lowercase hexadecimal digest")

    def to_dict(self) -> dict[str, Any]:
        return {
            "artifact_id": self.artifact_id,
            "run_id": self.run_id,
            "artifact_type": self.artifact_type,
            "mime_type": self.mime_type,
            "relative_path": self.relative_path,
            "sha256": self.sha256,
            "size_bytes": self.size_bytes,
            "producer_task_id": self.producer_task_id,
            "producer_tool": self.producer_tool,
            "provenance": _json_safe(self.provenance),
            "trust": self.trust.value,
            "created_at": self.created_at,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "ArtifactRecord":
        return cls(
            artifact_id=str(value["artifact_id"]),
            run_id=str(value["run_id"]),
            artifact_type=str(value["artifact_type"]),
            mime_type=str(value["mime_type"]),
            relative_path=str(value["relative_path"]),
            sha256=str(value["sha256"]),
            size_bytes=int(value["size_bytes"]),
            producer_task_id=(
                str(value["producer_task_id"]) if value.get("producer_task_id") else None
            ),
            producer_tool=str(value["producer_tool"]) if value.get("producer_tool") else None,
            provenance=dict(value.get("provenance") or {}),
            trust=ArtifactTrust(str(value.get("trust", ArtifactTrust.INTERNAL.value))),
            created_at=str(value.get("created_at") or _utc_now()),
        )

ArtifactTrust

Bases: str, Enum

Trust classification for artifact content.

Source code in src/cs_copilot/workflows/runtime.py

class ArtifactTrust(str, Enum):
    """Trust classification for artifact content."""

    INTERNAL = "internal"
    EXTERNAL = "external"
    UNTRUSTED = "untrusted"

EventReplayError

Bases: WorkflowRuntimeError

Raised when an event stream is malformed or cannot be replayed.

Source code in src/cs_copilot/workflows/runtime.py

class EventReplayError(WorkflowRuntimeError):
    """Raised when an event stream is malformed or cannot be replayed."""

HandoffEnvelope dataclass

Minimal, replayable context passed between workflow roles.

Source code in src/cs_copilot/workflows/runtime.py

@dataclass(frozen=True)
class HandoffEnvelope:
    """Minimal, replayable context passed between workflow roles."""

    handoff_id: str
    run_id: str
    workflow_slug: str
    task_id: str
    sender_role: str
    receiver_role: str
    objective: str
    trace_id: str
    span_id: str
    task_attempt: int | None = None
    parent_span_id: str | None = None
    constraints: tuple[str, ...] = ()
    required_capabilities: tuple[str, ...] = ()
    acceptance_criteria: tuple[str, ...] = ()
    input_artifact_ids: tuple[str, ...] = ()
    expected_output_artifacts: tuple[str, ...] = ()
    expected_output_schema: Mapping[str, Any] = field(default_factory=dict)
    context_summary: str | None = None
    budget: Mapping[str, Any] = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: _utc_now())
    schema_version: int = SCHEMA_VERSION

    def __post_init__(self) -> None:
        object.__setattr__(
            self, "handoff_id", validate_identifier(self.handoff_id, field="handoff_id")
        )
        object.__setattr__(self, "run_id", validate_identifier(self.run_id, field="run_id"))
        object.__setattr__(self, "task_id", validate_identifier(self.task_id, field="task_id"))
        object.__setattr__(self, "workflow_slug", sanitize_workflow_slug(self.workflow_slug))
        if self.schema_version != SCHEMA_VERSION:
            raise ValueError(
                f"unsupported handoff schema version {self.schema_version}; "
                f"expected {SCHEMA_VERSION}"
            )
        if self.task_attempt is not None and (
            isinstance(self.task_attempt, bool)
            or not isinstance(self.task_attempt, int)
            or self.task_attempt < 0
        ):
            raise ValueError("handoff task_attempt must be a non-negative integer")
        budget = self.budget.as_dict() if hasattr(self.budget, "as_dict") else self.budget
        if not isinstance(budget, Mapping):
            raise TypeError("handoff budget must be a mapping or expose as_dict()")
        object.__setattr__(self, "budget", dict(budget))
        output_schema = (
            self.expected_output_schema.as_dict()
            if hasattr(self.expected_output_schema, "as_dict")
            else self.expected_output_schema
        )
        if not isinstance(output_schema, Mapping):
            raise TypeError("handoff expected_output_schema must be a mapping")
        object.__setattr__(self, "expected_output_schema", dict(output_schema))
        _reject_private_handoff_context(
            {
                "expected_output_schema": output_schema,
                "budget": budget,
            }
        )
        for budget_name in ("max_tokens", "max_tool_calls", "timeout_seconds"):
            budget_value = budget.get(budget_name)
            if budget_value is not None and (
                isinstance(budget_value, bool)
                or not isinstance(budget_value, (int, float))
                or budget_value <= 0
            ):
                raise ValueError(f"handoff budget {budget_name} must be a positive number")
        for field_name in (
            "constraints",
            "required_capabilities",
            "acceptance_criteria",
            "input_artifact_ids",
            "expected_output_artifacts",
        ):
            object.__setattr__(self, field_name, _string_tuple(getattr(self, field_name)))
        required = (
            self.sender_role,
            self.receiver_role,
            self.objective,
            self.trace_id,
            self.span_id,
        )
        if any(not value.strip() for value in required):
            raise ValueError(
                "handoff sender, receiver, objective, trace_id, and span_id are required"
            )
        if self.sender_role == self.receiver_role:
            raise ValueError("handoff sender_role and receiver_role must differ")

    def to_dict(self) -> dict[str, Any]:
        return {
            "handoff_id": self.handoff_id,
            "run_id": self.run_id,
            "workflow_slug": self.workflow_slug,
            "task_id": self.task_id,
            "sender_role": self.sender_role,
            "receiver_role": self.receiver_role,
            "objective": self.objective,
            "task_attempt": self.task_attempt,
            "constraints": list(self.constraints),
            "required_capabilities": list(self.required_capabilities),
            "acceptance_criteria": list(self.acceptance_criteria),
            "input_artifact_ids": list(self.input_artifact_ids),
            "expected_output_artifacts": list(self.expected_output_artifacts),
            "expected_output_schema": _json_safe(self.expected_output_schema),
            "context_summary": self.context_summary,
            "budget": _json_safe(self.budget),
            "trace_id": self.trace_id,
            "span_id": self.span_id,
            "parent_span_id": self.parent_span_id,
            "created_at": self.created_at,
            "schema_version": self.schema_version,
        }

    @classmethod
    def create(
        cls,
        *,
        run_id: str,
        workflow_slug: str,
        task_id: str,
        sender_role: str,
        receiver_role: str,
        objective: str,
        trace_id: str | None = None,
        parent_span_id: str | None = None,
        **kwargs: Any,
    ) -> "HandoffEnvelope":
        """Create a handoff with generated stable and trace identifiers."""

        return cls(
            handoff_id=f"handoff-{uuid.uuid4().hex[:12]}",
            run_id=run_id,
            workflow_slug=workflow_slug,
            task_id=task_id,
            sender_role=sender_role,
            receiver_role=receiver_role,
            objective=objective,
            trace_id=trace_id or uuid.uuid4().hex,
            span_id=uuid.uuid4().hex,
            parent_span_id=parent_span_id,
            **kwargs,
        )

    @classmethod
    def from_mapping(cls, value: Mapping[str, Any]) -> "HandoffEnvelope":
        """Validate an external handoff and stamp its creation time locally."""

        _reject_private_handoff_context(value)
        return cls._from_mapping_fields(value, created_at=_utc_now())

    @classmethod
    def _from_mapping_fields(
        cls,
        value: Mapping[str, Any],
        *,
        created_at: str,
    ) -> "HandoffEnvelope":
        return cls(
            handoff_id=str(value.get("handoff_id") or f"handoff-{uuid.uuid4().hex[:12]}"),
            run_id=str(value["run_id"]),
            workflow_slug=str(value["workflow_slug"]),
            task_id=str(value["task_id"]),
            sender_role=str(value["sender_role"]),
            receiver_role=str(value["receiver_role"]),
            objective=str(value["objective"]),
            task_attempt=(
                int(value["task_attempt"]) if value.get("task_attempt") is not None else None
            ),
            constraints=tuple(str(item) for item in value.get("constraints", ())),
            required_capabilities=tuple(
                str(item) for item in value.get("required_capabilities", ())
            ),
            acceptance_criteria=tuple(str(item) for item in value.get("acceptance_criteria", ())),
            input_artifact_ids=tuple(
                str(item) for item in value.get("input_artifact_ids", value.get("artifact_ids", ()))
            ),
            expected_output_artifacts=tuple(
                str(item) for item in value.get("expected_output_artifacts", ())
            ),
            expected_output_schema=dict(value.get("expected_output_schema") or {}),
            context_summary=(
                str(value["context_summary"]) if value.get("context_summary") else None
            ),
            budget=dict(value.get("budget") or {}),
            trace_id=str(value.get("trace_id") or ""),
            span_id=str(value.get("span_id") or ""),
            parent_span_id=(str(value["parent_span_id"]) if value.get("parent_span_id") else None),
            created_at=created_at,
            schema_version=int(value.get("schema_version", SCHEMA_VERSION)),
        )

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "HandoffEnvelope":
        """Restore a persisted handoff, including its authoritative timestamp."""

        _reject_private_handoff_context(value)
        return cls._from_mapping_fields(
            value,
            created_at=str(value.get("created_at") or _utc_now()),
        )

create(*, run_id, workflow_slug, task_id, sender_role, receiver_role, objective, trace_id=None, parent_span_id=None, **kwargs) classmethod

Create a handoff with generated stable and trace identifiers.

Source code in src/cs_copilot/workflows/runtime.py

@classmethod
def create(
    cls,
    *,
    run_id: str,
    workflow_slug: str,
    task_id: str,
    sender_role: str,
    receiver_role: str,
    objective: str,
    trace_id: str | None = None,
    parent_span_id: str | None = None,
    **kwargs: Any,
) -> "HandoffEnvelope":
    """Create a handoff with generated stable and trace identifiers."""

    return cls(
        handoff_id=f"handoff-{uuid.uuid4().hex[:12]}",
        run_id=run_id,
        workflow_slug=workflow_slug,
        task_id=task_id,
        sender_role=sender_role,
        receiver_role=receiver_role,
        objective=objective,
        trace_id=trace_id or uuid.uuid4().hex,
        span_id=uuid.uuid4().hex,
        parent_span_id=parent_span_id,
        **kwargs,
    )

from_mapping(value) classmethod

Validate an external handoff and stamp its creation time locally.

Source code in src/cs_copilot/workflows/runtime.py

@classmethod
def from_mapping(cls, value: Mapping[str, Any]) -> "HandoffEnvelope":
    """Validate an external handoff and stamp its creation time locally."""

    _reject_private_handoff_context(value)
    return cls._from_mapping_fields(value, created_at=_utc_now())

from_dict(value) classmethod

Restore a persisted handoff, including its authoritative timestamp.

Source code in src/cs_copilot/workflows/runtime.py

@classmethod
def from_dict(cls, value: Mapping[str, Any]) -> "HandoffEnvelope":
    """Restore a persisted handoff, including its authoritative timestamp."""

    _reject_private_handoff_context(value)
    return cls._from_mapping_fields(
        value,
        created_at=str(value.get("created_at") or _utc_now()),
    )

InvalidTransitionError

Bases: WorkflowRuntimeError

Raised when a run or task lifecycle transition is illegal.

Source code in src/cs_copilot/workflows/runtime.py

class InvalidTransitionError(WorkflowRuntimeError):
    """Raised when a run or task lifecycle transition is illegal."""

RunContext

Event-sourced workflow runtime bound to one session-scoped run.

Source code in src/cs_copilot/workflows/runtime.py

class RunContext:
    """Event-sourced workflow runtime bound to one session-scoped run."""

    def __init__(
        self,
        *,
        layout: OutputLayout,
        run: WorkflowRun | None = None,
        events: Sequence[WorkflowEvent] = (),
    ) -> None:
        self.layout = layout
        self.run = run
        self.events = list(events)
        self._event_lock = threading.RLock()
        self._syncs_since_full = 0

    @classmethod
    def create(
        cls,
        workflow_slug: str,
        *,
        session_state: dict[str, Any] | None = None,
        session_id: str | None = None,
        run_id: str | None = None,
        constraints: Mapping[str, Any] | None = None,
        budget: Mapping[str, Any] | None = None,
        trace_id: str | None = None,
    ) -> "RunContext":
        slug = sanitize_workflow_slug(workflow_slug)
        if session_id is not None:
            _activate_session(session_id)
        selected_run_id = run_id or f"{slug}-{uuid.uuid4().hex[:12]}"
        output_context = ensure_output_context(
            session_state,
            workflow_slug=slug,
            run_id=selected_run_id,
            session_id=session_id,
        )
        run = WorkflowRun(
            session_id=str(output_context["session_id"]),
            run_id=str(output_context["run_id"]),
            workflow_slug=str(output_context["workflow_slug"]),
            trace_id=trace_id or uuid.uuid4().hex,
            workflow_contract=_workflow_contract_snapshot(slug),
            constraints=dict(constraints or {}),
            budget=dict(budget or {}),
        )
        context = cls(
            layout=OutputLayout(run.session_id, run.run_id, run.workflow_slug),
        )
        context.append_event("run_created", {"run": run.to_dict()})
        context.bind_session_state(session_state)
        return context

    @classmethod
    def load(
        cls,
        run_id: str,
        *,
        session_id: str | None = None,
        verify_artifacts: bool = False,
    ) -> "RunContext":
        run_id = validate_identifier(run_id, field="run_id")
        resolved_session = validate_identifier(
            session_id or _current_session_id(), field="session_id"
        )
        _activate_session(resolved_session)
        provisional = OutputLayout(resolved_session, run_id, "workflow")
        events = _read_events(provisional)
        run = _replay(events)
        if run.session_id != resolved_session:
            raise EventReplayError(
                f"run {run_id!r} belongs to session {run.session_id!r}, not {resolved_session!r}"
            )
        context = cls(
            layout=OutputLayout(run.session_id, run.run_id, run.workflow_slug),
            run=run,
            events=events,
        )
        if verify_artifacts:
            context.verify_artifacts()
        return context

    @classmethod
    def resume(cls, run_id: str, *, session_id: str | None = None) -> "RunContext":
        """Load a run and verify all registered inputs before more work is scheduled."""

        return cls.load(run_id, session_id=session_id, verify_artifacts=True)

    @classmethod
    def from_session_state(
        cls,
        session_state: Mapping[str, Any],
        *,
        verify_artifacts: bool = False,
    ) -> "RunContext":
        """Load the run named by canonical ``session_state`` output context."""

        output_context = session_state.get(OUTPUT_CONTEXT_KEY)
        if not isinstance(output_context, Mapping):
            raise ValueError(f"session_state[{OUTPUT_CONTEXT_KEY!r}] is missing")
        return cls.load(
            str(output_context["run_id"]),
            session_id=str(output_context["session_id"]),
            verify_artifacts=verify_artifacts,
        )

    def bind_session_state(
        self,
        session_state: dict[str, Any] | None,
        *,
        span_id: str | None = None,
        parent_span_id: str | None = None,
    ) -> dict[str, Any]:
        """Expose serializable run identity and trace context to agent/tool code."""

        run = self._require_run()
        output_context = {
            "layout_version": LAYOUT_VERSION,
            "session_id": run.session_id,
            "run_id": run.run_id,
            "workflow_slug": run.workflow_slug,
            "trace_id": run.trace_id,
            "span_id": span_id or run.trace_id,
            "parent_span_id": parent_span_id,
        }
        if isinstance(session_state, dict):
            session_state[OUTPUT_CONTEXT_KEY] = output_context
        return output_context

    def append_event(
        self,
        event_type: str,
        payload: Mapping[str, Any] | None = None,
        *,
        precondition: (
            Callable[
                [WorkflowRun | None, Sequence[WorkflowEvent]],
                None,
            ]
            | None
        ) = None,
    ) -> WorkflowEvent:
        """Validate, persist, and apply one immutable event.

        Snapshots are refreshed on a best-effort basis after the immutable
        event commits. A replaceable snapshot failure must never report the
        committed event as failed.
        """

        if not event_type.strip():
            raise ValueError("event_type cannot be empty")
        with self._event_lock, _run_lock(self.layout):
            for attempt in range(3):
                self._synchronize_events(pending_event_type=event_type)
                if precondition is not None:
                    precondition(self.run, self.events)
                sequence = len(self.events) + 1
                event = WorkflowEvent(
                    # A deterministic name reserves the sequence atomically.
                    # The run id scopes event identity, so a UUID is unnecessary.
                    event_id=f"{sequence:08d}",
                    sequence=sequence,
                    run_id=self.layout.run_id,
                    event_type=event_type,
                    timestamp=_utc_now(),
                    payload=dict(payload or {}),
                )
                # Apply to a detached copy so a failed immutable write cannot
                # mutate the caller-visible snapshot.
                base = WorkflowRun.from_dict(self.run.to_dict()) if self.run is not None else None
                candidate = _apply_event(base, event)
                try:
                    _write_event(self.layout, event)
                except _EventConflictError as exc:
                    if attempt == 2:
                        raise WorkflowRuntimeError(
                            f"could not reserve the next event for run "
                            f"{self.layout.run_id!r} after concurrent writes"
                        ) from exc
                    continue
                self.events.append(event)
                self.run = candidate
                if event_type in _OBSERVATION_EVENT_TYPES:
                    # Observational events never change run state; rewriting
                    # both replaceable snapshots for each of them would double
                    # the storage writes of every tool call. The next
                    # state-changing event (or rebuild_snapshots) refreshes them.
                    return event
                try:
                    self._write_snapshots()
                except Exception:
                    logger.warning(
                        "Workflow event %s committed for run %s, but derived "
                        "snapshot refresh failed",
                        event.event_id,
                        self.layout.run_id,
                        exc_info=True,
                    )
                return event
        raise AssertionError("unreachable event append state")  # pragma: no cover

    def _synchronize_events(self, *, pending_event_type: str) -> None:
        """Refresh a stale writer and reject replaced or duplicate run streams.

        A context that already holds the stream catches up incrementally: it
        checks that its last known event is unchanged and reads newer events by
        their exact sequence names, so an append costs neither a directory
        listing nor a replay of the known prefix. Every
        ``_STREAM_CHECK_INTERVAL`` catch-ups, one listing confirms that no known
        event disappeared. A full re-read and replay runs for a context without
        events and whenever a known event is missing.
        """

        if self.events and self.run is not None:
            known_prefix_listed = True
            if self._syncs_since_full >= _STREAM_CHECK_INTERVAL:
                known_prefix_listed = self._known_events_listed()
                self._syncs_since_full = 0
            if known_prefix_listed and self._synchronize_tail():
                self._syncs_since_full += 1
                return
        self._synchronize_full(pending_event_type=pending_event_type)
        self._syncs_since_full = 0

    def _known_events_listed(self) -> bool:
        """Return whether every event this context knows is still in the stream."""

        listed = {PurePosixPath(path).name for path in _list_event_paths(self.layout)}
        return all(f"{event.event_id}.jsonl" in listed for event in self.events)

    def _synchronize_tail(self) -> bool:
        """Catch up from the known tail; return ``False`` when a full re-read is needed."""

        _forget_cached_event_listing(self.layout)
        last = self.events[-1]
        try:
            stored = _read_event_segment(self.layout, last.event_id)
        except FileNotFoundError:
            return False
        if _canonical_event(stored) != _canonical_event(last):
            raise EventReplayError(f"event stream for run {self.layout.run_id!r} was replaced")

        newer: list[WorkflowEvent] = []
        sequence = last.sequence + 1
        while True:
            try:
                event = _read_event_segment(self.layout, f"{sequence:08d}")
            except FileNotFoundError:
                break
            if event.sequence != sequence:
                raise EventReplayError(f"event sequence is not contiguous at {sequence}")
            newer.append(event)
            sequence += 1
        if newer:
            # Apply to a detached copy so a malformed event cannot leave the
            # caller-visible run half updated.
            run = WorkflowRun.from_dict(self._require_run().to_dict())
            for event in newer:
                run = _apply_replayed_event(run, event)
            self.events.extend(newer)
            self.run = run
        return True

    def _synchronize_full(self, *, pending_event_type: str) -> None:
        paths = _list_event_paths(self.layout)
        if not paths:
            if self.events or self.run is not None:
                raise EventReplayError(f"event stream for run {self.layout.run_id!r} disappeared")
            return

        authoritative = _read_events(self.layout)
        if not self.events:
            if pending_event_type == "run_created":
                raise WorkflowRuntimeError(f"workflow run {self.layout.run_id!r} already exists")
            if self.run is None:
                raise EventReplayError(
                    f"run {self.layout.run_id!r} has events but no local run state"
                )
        else:
            known_ids = [event.event_id for event in self.events]
            authoritative_prefix = [event.event_id for event in authoritative[: len(known_ids)]]
            if authoritative_prefix != known_ids:
                raise EventReplayError(f"event stream for run {self.layout.run_id!r} was replaced")
        self.events = authoritative
        self.run = _replay(authoritative)

    def refresh(self, *, verify_artifacts: bool = False) -> WorkflowRun:
        """Refresh this context from its authoritative event stream."""

        with _run_lock(self.layout):
            self._synchronize_events(pending_event_type="refresh")
            run = self._require_run()
            if verify_artifacts:
                self.verify_artifacts()
            return run

    def pending_tool_invocations(
        self,
        *,
        task_id: str | None = None,
        domain_only: bool = False,
    ) -> tuple[str, ...]:
        """Return authoritative MCP calls that have not reached a terminal event."""

        self.refresh()
        return tuple(
            _pending_tool_invocations(
                self.events,
                task_id=task_id,
                authoritative_run=self.run,
                domain_only=domain_only,
            )
        )

    def abandon_tool_invocation(
        self,
        span_id: str,
        *,
        reason: str,
    ) -> WorkflowEvent:
        """Reconcile one confirmed-orphaned MCP invocation after a crash."""

        selected_span = validate_identifier(span_id, field="span_id")
        selected_reason = str(reason).strip()
        if not selected_reason:
            raise ValueError("orphaned invocation recovery requires a reason")
        if len(selected_reason) > 1000:
            raise ValueError("orphaned invocation recovery reason exceeds 1000 characters")
        self.refresh()
        started = next(
            (
                event.payload
                for event in reversed(self.events)
                if event.event_type == "tool_progress"
                and event.payload.get("stage") == "started"
                and str(event.payload.get("span_id") or "") == selected_span
            ),
            None,
        )
        if started is None or not _is_pending_tool_span(
            self.events,
            selected_span,
            authoritative_run=self.run,
        ):
            raise InvalidTransitionError(f"MCP invocation {selected_span!r} is not pending")
        tool_name = str(started.get("tool_name") or "")
        if not tool_name or tool_name.startswith("workflow_"):
            raise InvalidTransitionError("only orphaned domain-tool invocations may be reconciled")
        payload = {
            key: started.get(key)
            for key in (
                "runtime",
                "session_id",
                "run_id",
                "workflow_slug",
                "trace_id",
                "span_id",
                "parent_span_id",
                "tool_name",
                "task_id",
                "role",
                "profile",
                "attempt",
                "max_attempts",
                "task_attempt",
                "handoff_id",
            )
        }
        payload.update(
            {
                "stage": "abandoned",
                "cached": False,
                "message": selected_reason,
                "recovery": {
                    "confirmed_not_running": True,
                    "reason": selected_reason,
                },
            }
        )

        def require_still_orphaned(
            authoritative_run: WorkflowRun | None,
            events: Sequence[WorkflowEvent],
        ) -> None:
            if not _is_pending_tool_span(
                events,
                selected_span,
                authoritative_run=authoritative_run,
            ):
                raise InvalidTransitionError(
                    f"MCP invocation {selected_span!r} is no longer pending"
                )

        return self.append_event(
            "tool_progress",
            payload,
            precondition=require_still_orphaned,
        )

    def transition_run(
        self,
        status: RunStatus | str,
        *,
        reason: str | None = None,
        error: ToolError | None = None,
    ) -> WorkflowRun:
        target = RunStatus(status)
        if target is RunStatus.FAILED and error is None:
            raise ValueError("failed run transition requires a structured error")
        if error is not None and target is not RunStatus.FAILED:
            raise ValueError("run errors may only accompany a failed transition")
        run = self._require_run()
        if target is RunStatus.COMPLETED and run.status is RunStatus.RUNNING:
            self.verify_artifacts()
            reasons = _completion_gaps(run)
            if reasons:
                raise InvalidTransitionError(
                    "cannot mark workflow completed before its contracts are "
                    f"satisfied: {'; '.join(reasons)}"
                )
        payload: dict[str, Any] = {"status": target.value}
        if reason:
            payload["reason"] = reason
        if error:
            payload["error"] = error.to_dict()

        def reject_inflight_calls(
            authoritative_run: WorkflowRun | None,
            events: Sequence[WorkflowEvent],
        ) -> None:
            pending = _pending_tool_invocations(
                events,
                authoritative_run=authoritative_run,
                domain_only=True,
            )
            if pending:
                raise InvalidTransitionError(
                    "workflow status cannot change while MCP tool calls are in "
                    f"flight: {', '.join(pending)}"
                )

        self.append_event(
            "run_status_changed",
            payload,
            precondition=reject_inflight_calls,
        )
        return self._require_run()

    def add_task(self, task: TaskRecord) -> TaskRecord:
        run = self._require_run()
        _require_mutable_run(run, operation="add a task")
        if task.task_id in run.tasks:
            raise ValueError(f"duplicate task_id: {task.task_id}")
        if task.status is not TaskStatus.PENDING:
            raise ValueError("new tasks must start in pending status")
        if task.attempts != 0 or task.output_artifact_ids or task.error is not None:
            raise ValueError("new tasks cannot contain attempts, outputs, or an error")
        contract = _run_task_contract(run, task.task_id)
        if run.workflow_contract.get("tasks") and contract is None:
            raise ValueError(
                f"task {task.task_id!r} is not declared by workflow " f"{run.workflow_slug!r}"
            )
        if contract is not None:
            required_role = str(contract["role"])
            required_profile = str(contract["profile"])
            if task.role != required_role:
                raise ValueError(
                    f"task {task.task_id!r} requires role {required_role!r}, " f"not {task.role!r}"
                )
            if task.profile != required_profile:
                raise ValueError(
                    f"task {task.task_id!r} requires profile {required_profile!r}, "
                    f"not {task.profile!r}"
                )
        unknown_inputs = sorted(set(task.input_artifact_ids) - set(run.artifacts))
        if unknown_inputs:
            raise ValueError(
                f"task references unknown input artifacts: {', '.join(unknown_inputs)}"
            )
        self.append_event("task_created", {"task": task.to_dict()})
        return self._require_run().tasks[task.task_id]

    def transition_task(
        self,
        task_id: str,
        status: TaskStatus | str,
        *,
        error: ToolError | None = None,
    ) -> TaskRecord:
        task_id = validate_identifier(task_id, field="task_id")
        target = TaskStatus(status)
        if target is TaskStatus.FAILED and error is None:
            raise ValueError("failed task transition requires a structured error")
        if error is not None and target is not TaskStatus.FAILED:
            raise ValueError("task errors may only accompany a failed transition")
        run = self._require_run()
        _require_mutable_run(run, operation="transition a task")
        if target is TaskStatus.RUNNING:
            if run.status is not RunStatus.RUNNING:
                raise InvalidTransitionError(
                    f"task {task_id!r} cannot run while workflow status is " f"{run.status.value!r}"
                )
            contract = _run_task_contract(run, task_id)
            if contract is not None:
                incomplete = [
                    dependency
                    for dependency in contract.get("depends_on", ())
                    if dependency not in run.tasks
                    or run.tasks[dependency].status is not TaskStatus.COMPLETED
                ]
                if incomplete:
                    raise InvalidTransitionError(
                        f"task {task_id!r} cannot run before dependencies complete: "
                        + ", ".join(incomplete)
                    )
                input_gaps = _task_input_gaps(run, task_id)
                if input_gaps:
                    raise InvalidTransitionError(
                        f"task {task_id!r} cannot run before required input "
                        f"artifacts are registered: {'; '.join(input_gaps)}"
                    )
                task = run.tasks.get(task_id)
                handoff_gap = _task_handoff_gap(run, task) if task is not None else None
                if handoff_gap:
                    raise InvalidTransitionError(handoff_gap)
            self.verify_task_inputs(task_id)
        if target is TaskStatus.COMPLETED:
            required_outputs = _run_required_task_outputs(run, task_id)
            produced = {
                artifact.artifact_type
                for artifact in run.artifacts.values()
                if artifact.producer_task_id == task_id
            }
            missing_outputs = sorted(set(required_outputs) - produced)
            if missing_outputs:
                raise InvalidTransitionError(
                    f"task {task_id!r} cannot complete before required artifacts "
                    f"are registered: {', '.join(missing_outputs)}"
                )
        payload: dict[str, Any] = {"task_id": task_id, "status": target.value}
        if error:
            payload["error"] = error.to_dict()

        def reject_inflight_calls(
            authoritative_run: WorkflowRun | None,
            events: Sequence[WorkflowEvent],
        ) -> None:
            authoritative_task = (
                authoritative_run.tasks.get(task_id) if authoritative_run is not None else None
            )
            selected_attempt = (
                int(authoritative_task.attempts) + (1 if target is TaskStatus.RUNNING else 0)
                if authoritative_task is not None
                else None
            )
            pending = _pending_tool_invocations(
                events,
                task_id=task_id,
                task_attempt=selected_attempt,
                domain_only=True,
            )
            if pending:
                raise InvalidTransitionError(
                    f"task {task_id!r} cannot change status while MCP tool calls "
                    f"are in flight: {', '.join(pending)}"
                )

        self.append_event(
            "task_status_changed",
            payload,
            precondition=reject_inflight_calls,
        )
        return self._require_run().tasks[task_id]

    def record_handoff(
        self,
        envelope: Any,
        *,
        input_artifact_contracts: Sequence[str] | None = None,
    ) -> HandoffEnvelope:
        """Persist a runtime or duck-typed agent handoff without importing agents."""

        if isinstance(envelope, HandoffEnvelope):
            handoff = envelope
        else:
            if hasattr(envelope, "to_dict"):
                value = envelope.to_dict()
            elif hasattr(envelope, "as_dict"):
                value = envelope.as_dict()
            elif isinstance(envelope, Mapping):
                value = envelope
            else:
                raise TypeError("handoff must be a mapping or expose to_dict()/as_dict()")
            handoff = HandoffEnvelope.from_dict(value)
        handoff = replace(handoff, created_at=_utc_now())
        run = self._require_run()
        _require_mutable_run(run, operation="record a handoff")
        if any(item.handoff_id == handoff.handoff_id for item in run.handoffs):
            raise ValueError(f"duplicate handoff_id: {handoff.handoff_id}")
        handoff = _validate_and_normalize_handoff(
            run,
            handoff,
            input_artifact_contracts=input_artifact_contracts,
        )
        for artifact_id in handoff.input_artifact_ids:
            self.verify_artifact(artifact_id)
        self.append_event("handoff_recorded", {"handoff": handoff.to_dict()})
        return self._require_run().handoffs[-1]

    def record_progress(
        self,
        *,
        task_id: str,
        current: int | float,
        total: int | float | None = None,
        message: str | None = None,
    ) -> WorkflowEvent:
        run = self._require_run()
        _require_mutable_run(run, operation="record task progress")
        selected_task_id = validate_identifier(task_id, field="task_id")
        if selected_task_id not in run.tasks:
            raise ValueError(f"progress references unknown task {selected_task_id!r}")
        if (
            isinstance(current, bool)
            or not isinstance(current, (int, float))
            or not math.isfinite(current)
            or current < 0
        ):
            raise ValueError("task progress current must be a finite non-negative number")
        if total is not None and (
            isinstance(total, bool)
            or not isinstance(total, (int, float))
            or not math.isfinite(total)
            or total <= 0
        ):
            raise ValueError("task progress total must be a finite positive number")
        if total is not None and current > total:
            raise ValueError("task progress current cannot exceed total")
        payload: dict[str, Any] = {
            "task_id": selected_task_id,
            "current": current,
        }
        if total is not None:
            payload["total"] = total
        if message:
            payload["message"] = message
        return self.append_event("task_progress", payload)

    def register_artifact(
        self,
        path: str,
        *,
        artifact_type: str,
        mime_type: str,
        artifact_id: str | None = None,
        producer_task_id: str | None = None,
        active_task_id: str | None = None,
        producer_tool: str | None = None,
        provenance: Mapping[str, Any] | None = None,
        trust: ArtifactTrust | str | None = None,
    ) -> ArtifactRecord:
        """Checksum and register a file that is contained by this run's root."""

        run = self._require_run()
        _require_mutable_run(run, operation="register an artifact")
        if active_task_id is not None:
            active_task_id = validate_identifier(active_task_id, field="active_task_id")
        if producer_task_id is not None:
            producer_task_id = validate_identifier(
                producer_task_id,
                field="producer_task_id",
            )
            try:
                producer_task = run.tasks[producer_task_id]
            except KeyError as exc:
                raise ValueError(f"artifact references unknown task {producer_task_id!r}") from exc
            if _run_task_contract(run, producer_task_id) is not None:
                if active_task_id != producer_task_id:
                    raise ValueError(
                        f"producer_task_id {producer_task_id!r} does not match the "
                        f"active task {active_task_id!r}"
                    )
                if producer_task.status is not TaskStatus.RUNNING:
                    raise ValueError(
                        f"producer task {producer_task_id!r} must be an active "
                        "RUNNING catalog task"
                    )
        relative = _normalize_artifact_input_path(self.layout, path)
        digest, size = _digest_artifact(self.layout, relative)
        existing = next(
            (artifact for artifact in run.artifacts.values() if artifact.relative_path == relative),
            None,
        )
        if existing is not None:
            if digest != existing.sha256 or size != existing.size_bytes:
                raise ArtifactIntegrityError(
                    f"registered artifact path {relative!r} changed after registration"
                )
            if artifact_id is not None and artifact_id != existing.artifact_id:
                raise ValueError(
                    f"artifact path {relative!r} is already registered as "
                    f"{existing.artifact_id!r}"
                )
            if artifact_type != existing.artifact_type or mime_type != existing.mime_type:
                raise ValueError(
                    f"artifact path {relative!r} is already registered with "
                    "different type or MIME metadata"
                )
            if producer_task_id is not None and producer_task_id != existing.producer_task_id:
                raise ValueError(
                    f"artifact path {relative!r} is already registered to a "
                    "different producer task"
                )
            if producer_tool is not None and producer_tool != existing.producer_tool:
                raise ValueError(
                    f"artifact path {relative!r} is already registered to a "
                    "different producer tool"
                )
            if provenance is not None and _json_safe(provenance) != _json_safe(existing.provenance):
                raise ValueError(
                    f"artifact path {relative!r} is already registered with " "different provenance"
                )
            if trust is not None and ArtifactTrust(trust) is not existing.trust:
                raise ValueError(
                    f"artifact path {relative!r} is already registered with "
                    "a different trust classification"
                )
            return existing
        selected_id = artifact_id or f"artifact-{uuid.uuid4().hex[:12]}"
        record = ArtifactRecord(
            artifact_id=selected_id,
            run_id=run.run_id,
            artifact_type=artifact_type,
            mime_type=mime_type,
            relative_path=relative,
            sha256=digest,
            size_bytes=size,
            producer_task_id=producer_task_id,
            producer_tool=producer_tool,
            provenance=dict(provenance or {}),
            trust=ArtifactTrust(trust or ArtifactTrust.INTERNAL),
        )
        if record.artifact_id in run.artifacts:
            raise ValueError(f"duplicate artifact_id: {record.artifact_id}")
        existing_input_id = run.workflow_inputs.get(record.artifact_type)
        if existing_input_id is not None and existing_input_id != record.artifact_id:
            raise ValueError(
                f"workflow input contract {record.artifact_type!r} is already bound "
                f"to artifact {existing_input_id!r}"
            )
        if producer_task_id:
            task_contract = _run_task_contract(run, producer_task_id)
            if task_contract is not None and artifact_type not in set(
                task_contract.get("output_artifacts", ())
            ):
                raise ValueError(
                    f"task {producer_task_id!r} is not contracted to produce "
                    f"artifact type {artifact_type!r}"
                )

        def authorize_registration(
            authoritative_run: WorkflowRun | None,
            events: Sequence[WorkflowEvent],
        ) -> None:
            _authorize_artifact_registration(
                authoritative_run,
                events,
                record=record,
                active_task_id=active_task_id,
            )

        self.append_event(
            "artifact_registered",
            {"artifact": record.to_dict()},
            precondition=authorize_registration,
        )
        return self._require_run().artifacts[record.artifact_id]

    def verify_artifact(self, artifact_id: str) -> ArtifactRecord:
        run = self._require_run()
        try:
            record = run.artifacts[artifact_id]
        except KeyError as exc:
            raise KeyError(f"unknown artifact_id: {artifact_id}") from exc
        try:
            digest, size = _digest_artifact(self.layout, record.relative_path)
        except FileNotFoundError as exc:
            raise ArtifactIntegrityError(
                f"artifact {artifact_id!r} is missing: {record.relative_path}"
            ) from exc
        if digest != record.sha256 or size != record.size_bytes:
            raise ArtifactIntegrityError(
                f"artifact {artifact_id!r} failed checksum/size verification"
            )
        return record

    def verify_artifacts(self) -> list[ArtifactRecord]:
        return [self.verify_artifact(key) for key in sorted(self._require_run().artifacts)]

    def verify_task_inputs(self, task_id: str) -> list[ArtifactRecord]:
        """Verify the selected input artifacts for one task before execution."""

        selected_task_id = validate_identifier(task_id, field="task_id")
        run = self._require_run()
        try:
            task = run.tasks[selected_task_id]
        except KeyError as exc:
            raise ValueError(f"unknown task_id: {selected_task_id}") from exc
        input_gaps = _task_input_gaps(run, selected_task_id)
        if input_gaps:
            raise InvalidTransitionError(
                f"task {selected_task_id!r} has invalid required input "
                f"artifacts: {'; '.join(input_gaps)}"
            )
        return [
            self.verify_artifact(artifact_id)
            for artifact_id in dict.fromkeys(task.input_artifact_ids)
        ]

    def rebuild_snapshots(self) -> WorkflowRun:
        """Replay immutable events and replace both derived snapshots."""

        events = _read_events(self.layout)
        run = _replay(events)
        self.events = events
        self.run = run
        self._syncs_since_full = 0
        self.layout = OutputLayout(run.session_id, run.run_id, run.workflow_slug)
        self._write_snapshots()
        return run

    def complete(
        self,
        *,
        required_artifact_types: Sequence[str] | None = None,
        required_task_ids: Sequence[str] | None = None,
    ) -> WorkflowRun:
        """Complete the run only after its catalog contracts are satisfied.

        Catalog requirements are authoritative and explicit requirements can
        only make them stricter. Ad-hoc runs require every registered task and
        any explicitly requested artifacts to be complete.
        """

        run = self._require_run()
        if run.status in {RunStatus.COMPLETED, RunStatus.PARTIAL}:
            return run
        if run.status in {RunStatus.FAILED, RunStatus.CANCELLED}:
            raise InvalidTransitionError(
                f"cannot complete terminal run in status {run.status.value!r}"
            )
        if run.status is not RunStatus.RUNNING:
            run = self.transition_run(
                RunStatus.RUNNING,
                reason="completion validation requested",
            )
        self.verify_artifacts()
        reasons = _completion_gaps(
            run,
            required_artifact_types=required_artifact_types,
            required_task_ids=required_task_ids,
        )
        if reasons:
            return self.transition_run(RunStatus.PARTIAL, reason="; ".join(reasons))
        return self.transition_run(RunStatus.COMPLETED)

    def manifest_payload(self) -> dict[str, Any]:
        run = self._require_run()
        payload = run.to_dict()
        payload["event_count"] = len(self.events)
        payload["last_event_id"] = self.events[-1].event_id if self.events else None
        return payload

    def artifact_index_payload(self) -> dict[str, Any]:
        run = self._require_run()
        return {
            "schema_version": SCHEMA_VERSION,
            "session_id": run.session_id,
            "run_id": run.run_id,
            "workflow_slug": run.workflow_slug,
            "artifacts": [run.artifacts[key].to_dict() for key in sorted(run.artifacts)],
        }

    def _write_snapshots(self) -> None:
        _write_json(self.layout.manifest_rel_path, self.manifest_payload())
        _write_json(self.layout.artifact_index_rel_path, self.artifact_index_payload())

    def _require_run(self) -> WorkflowRun:
        if self.run is None:
            raise WorkflowRuntimeError("workflow run has not been initialized")
        return self.run

resume(run_id, *, session_id=None) classmethod

Load a run and verify all registered inputs before more work is scheduled.

Source code in src/cs_copilot/workflows/runtime.py

@classmethod
def resume(cls, run_id: str, *, session_id: str | None = None) -> "RunContext":
    """Load a run and verify all registered inputs before more work is scheduled."""

    return cls.load(run_id, session_id=session_id, verify_artifacts=True)

from_session_state(session_state, *, verify_artifacts=False) classmethod

Load the run named by canonical session_state output context.

Source code in src/cs_copilot/workflows/runtime.py

@classmethod
def from_session_state(
    cls,
    session_state: Mapping[str, Any],
    *,
    verify_artifacts: bool = False,
) -> "RunContext":
    """Load the run named by canonical ``session_state`` output context."""

    output_context = session_state.get(OUTPUT_CONTEXT_KEY)
    if not isinstance(output_context, Mapping):
        raise ValueError(f"session_state[{OUTPUT_CONTEXT_KEY!r}] is missing")
    return cls.load(
        str(output_context["run_id"]),
        session_id=str(output_context["session_id"]),
        verify_artifacts=verify_artifacts,
    )

bind_session_state(session_state, *, span_id=None, parent_span_id=None)

Expose serializable run identity and trace context to agent/tool code.

Source code in src/cs_copilot/workflows/runtime.py

def bind_session_state(
    self,
    session_state: dict[str, Any] | None,
    *,
    span_id: str | None = None,
    parent_span_id: str | None = None,
) -> dict[str, Any]:
    """Expose serializable run identity and trace context to agent/tool code."""

    run = self._require_run()
    output_context = {
        "layout_version": LAYOUT_VERSION,
        "session_id": run.session_id,
        "run_id": run.run_id,
        "workflow_slug": run.workflow_slug,
        "trace_id": run.trace_id,
        "span_id": span_id or run.trace_id,
        "parent_span_id": parent_span_id,
    }
    if isinstance(session_state, dict):
        session_state[OUTPUT_CONTEXT_KEY] = output_context
    return output_context

append_event(event_type, payload=None, *, precondition=None)

Validate, persist, and apply one immutable event.

Snapshots are refreshed on a best-effort basis after the immutable event commits. A replaceable snapshot failure must never report the committed event as failed.

Source code in src/cs_copilot/workflows/runtime.py

def append_event(
    self,
    event_type: str,
    payload: Mapping[str, Any] | None = None,
    *,
    precondition: (
        Callable[
            [WorkflowRun | None, Sequence[WorkflowEvent]],
            None,
        ]
        | None
    ) = None,
) -> WorkflowEvent:
    """Validate, persist, and apply one immutable event.

    Snapshots are refreshed on a best-effort basis after the immutable
    event commits. A replaceable snapshot failure must never report the
    committed event as failed.
    """

    if not event_type.strip():
        raise ValueError("event_type cannot be empty")
    with self._event_lock, _run_lock(self.layout):
        for attempt in range(3):
            self._synchronize_events(pending_event_type=event_type)
            if precondition is not None:
                precondition(self.run, self.events)
            sequence = len(self.events) + 1
            event = WorkflowEvent(
                # A deterministic name reserves the sequence atomically.
                # The run id scopes event identity, so a UUID is unnecessary.
                event_id=f"{sequence:08d}",
                sequence=sequence,
                run_id=self.layout.run_id,
                event_type=event_type,
                timestamp=_utc_now(),
                payload=dict(payload or {}),
            )
            # Apply to a detached copy so a failed immutable write cannot
            # mutate the caller-visible snapshot.
            base = WorkflowRun.from_dict(self.run.to_dict()) if self.run is not None else None
            candidate = _apply_event(base, event)
            try:
                _write_event(self.layout, event)
            except _EventConflictError as exc:
                if attempt == 2:
                    raise WorkflowRuntimeError(
                        f"could not reserve the next event for run "
                        f"{self.layout.run_id!r} after concurrent writes"
                    ) from exc
                continue
            self.events.append(event)
            self.run = candidate
            if event_type in _OBSERVATION_EVENT_TYPES:
                # Observational events never change run state; rewriting
                # both replaceable snapshots for each of them would double
                # the storage writes of every tool call. The next
                # state-changing event (or rebuild_snapshots) refreshes them.
                return event
            try:
                self._write_snapshots()
            except Exception:
                logger.warning(
                    "Workflow event %s committed for run %s, but derived "
                    "snapshot refresh failed",
                    event.event_id,
                    self.layout.run_id,
                    exc_info=True,
                )
            return event
    raise AssertionError("unreachable event append state")  # pragma: no cover

refresh(*, verify_artifacts=False)

Refresh this context from its authoritative event stream.

Source code in src/cs_copilot/workflows/runtime.py

def refresh(self, *, verify_artifacts: bool = False) -> WorkflowRun:
    """Refresh this context from its authoritative event stream."""

    with _run_lock(self.layout):
        self._synchronize_events(pending_event_type="refresh")
        run = self._require_run()
        if verify_artifacts:
            self.verify_artifacts()
        return run

pending_tool_invocations(*, task_id=None, domain_only=False)

Return authoritative MCP calls that have not reached a terminal event.

Source code in src/cs_copilot/workflows/runtime.py

def pending_tool_invocations(
    self,
    *,
    task_id: str | None = None,
    domain_only: bool = False,
) -> tuple[str, ...]:
    """Return authoritative MCP calls that have not reached a terminal event."""

    self.refresh()
    return tuple(
        _pending_tool_invocations(
            self.events,
            task_id=task_id,
            authoritative_run=self.run,
            domain_only=domain_only,
        )
    )

abandon_tool_invocation(span_id, *, reason)

Reconcile one confirmed-orphaned MCP invocation after a crash.

Source code in src/cs_copilot/workflows/runtime.py

def abandon_tool_invocation(
    self,
    span_id: str,
    *,
    reason: str,
) -> WorkflowEvent:
    """Reconcile one confirmed-orphaned MCP invocation after a crash."""

    selected_span = validate_identifier(span_id, field="span_id")
    selected_reason = str(reason).strip()
    if not selected_reason:
        raise ValueError("orphaned invocation recovery requires a reason")
    if len(selected_reason) > 1000:
        raise ValueError("orphaned invocation recovery reason exceeds 1000 characters")
    self.refresh()
    started = next(
        (
            event.payload
            for event in reversed(self.events)
            if event.event_type == "tool_progress"
            and event.payload.get("stage") == "started"
            and str(event.payload.get("span_id") or "") == selected_span
        ),
        None,
    )
    if started is None or not _is_pending_tool_span(
        self.events,
        selected_span,
        authoritative_run=self.run,
    ):
        raise InvalidTransitionError(f"MCP invocation {selected_span!r} is not pending")
    tool_name = str(started.get("tool_name") or "")
    if not tool_name or tool_name.startswith("workflow_"):
        raise InvalidTransitionError("only orphaned domain-tool invocations may be reconciled")
    payload = {
        key: started.get(key)
        for key in (
            "runtime",
            "session_id",
            "run_id",
            "workflow_slug",
            "trace_id",
            "span_id",
            "parent_span_id",
            "tool_name",
            "task_id",
            "role",
            "profile",
            "attempt",
            "max_attempts",
            "task_attempt",
            "handoff_id",
        )
    }
    payload.update(
        {
            "stage": "abandoned",
            "cached": False,
            "message": selected_reason,
            "recovery": {
                "confirmed_not_running": True,
                "reason": selected_reason,
            },
        }
    )

    def require_still_orphaned(
        authoritative_run: WorkflowRun | None,
        events: Sequence[WorkflowEvent],
    ) -> None:
        if not _is_pending_tool_span(
            events,
            selected_span,
            authoritative_run=authoritative_run,
        ):
            raise InvalidTransitionError(
                f"MCP invocation {selected_span!r} is no longer pending"
            )

    return self.append_event(
        "tool_progress",
        payload,
        precondition=require_still_orphaned,
    )

record_handoff(envelope, *, input_artifact_contracts=None)

Persist a runtime or duck-typed agent handoff without importing agents.

Source code in src/cs_copilot/workflows/runtime.py

def record_handoff(
    self,
    envelope: Any,
    *,
    input_artifact_contracts: Sequence[str] | None = None,
) -> HandoffEnvelope:
    """Persist a runtime or duck-typed agent handoff without importing agents."""

    if isinstance(envelope, HandoffEnvelope):
        handoff = envelope
    else:
        if hasattr(envelope, "to_dict"):
            value = envelope.to_dict()
        elif hasattr(envelope, "as_dict"):
            value = envelope.as_dict()
        elif isinstance(envelope, Mapping):
            value = envelope
        else:
            raise TypeError("handoff must be a mapping or expose to_dict()/as_dict()")
        handoff = HandoffEnvelope.from_dict(value)
    handoff = replace(handoff, created_at=_utc_now())
    run = self._require_run()
    _require_mutable_run(run, operation="record a handoff")
    if any(item.handoff_id == handoff.handoff_id for item in run.handoffs):
        raise ValueError(f"duplicate handoff_id: {handoff.handoff_id}")
    handoff = _validate_and_normalize_handoff(
        run,
        handoff,
        input_artifact_contracts=input_artifact_contracts,
    )
    for artifact_id in handoff.input_artifact_ids:
        self.verify_artifact(artifact_id)
    self.append_event("handoff_recorded", {"handoff": handoff.to_dict()})
    return self._require_run().handoffs[-1]

register_artifact(path, *, artifact_type, mime_type, artifact_id=None, producer_task_id=None, active_task_id=None, producer_tool=None, provenance=None, trust=None)

Checksum and register a file that is contained by this run's root.

Source code in src/cs_copilot/workflows/runtime.py

def register_artifact(
    self,
    path: str,
    *,
    artifact_type: str,
    mime_type: str,
    artifact_id: str | None = None,
    producer_task_id: str | None = None,
    active_task_id: str | None = None,
    producer_tool: str | None = None,
    provenance: Mapping[str, Any] | None = None,
    trust: ArtifactTrust | str | None = None,
) -> ArtifactRecord:
    """Checksum and register a file that is contained by this run's root."""

    run = self._require_run()
    _require_mutable_run(run, operation="register an artifact")
    if active_task_id is not None:
        active_task_id = validate_identifier(active_task_id, field="active_task_id")
    if producer_task_id is not None:
        producer_task_id = validate_identifier(
            producer_task_id,
            field="producer_task_id",
        )
        try:
            producer_task = run.tasks[producer_task_id]
        except KeyError as exc:
            raise ValueError(f"artifact references unknown task {producer_task_id!r}") from exc
        if _run_task_contract(run, producer_task_id) is not None:
            if active_task_id != producer_task_id:
                raise ValueError(
                    f"producer_task_id {producer_task_id!r} does not match the "
                    f"active task {active_task_id!r}"
                )
            if producer_task.status is not TaskStatus.RUNNING:
                raise ValueError(
                    f"producer task {producer_task_id!r} must be an active "
                    "RUNNING catalog task"
                )
    relative = _normalize_artifact_input_path(self.layout, path)
    digest, size = _digest_artifact(self.layout, relative)
    existing = next(
        (artifact for artifact in run.artifacts.values() if artifact.relative_path == relative),
        None,
    )
    if existing is not None:
        if digest != existing.sha256 or size != existing.size_bytes:
            raise ArtifactIntegrityError(
                f"registered artifact path {relative!r} changed after registration"
            )
        if artifact_id is not None and artifact_id != existing.artifact_id:
            raise ValueError(
                f"artifact path {relative!r} is already registered as "
                f"{existing.artifact_id!r}"
            )
        if artifact_type != existing.artifact_type or mime_type != existing.mime_type:
            raise ValueError(
                f"artifact path {relative!r} is already registered with "
                "different type or MIME metadata"
            )
        if producer_task_id is not None and producer_task_id != existing.producer_task_id:
            raise ValueError(
                f"artifact path {relative!r} is already registered to a "
                "different producer task"
            )
        if producer_tool is not None and producer_tool != existing.producer_tool:
            raise ValueError(
                f"artifact path {relative!r} is already registered to a "
                "different producer tool"
            )
        if provenance is not None and _json_safe(provenance) != _json_safe(existing.provenance):
            raise ValueError(
                f"artifact path {relative!r} is already registered with " "different provenance"
            )
        if trust is not None and ArtifactTrust(trust) is not existing.trust:
            raise ValueError(
                f"artifact path {relative!r} is already registered with "
                "a different trust classification"
            )
        return existing
    selected_id = artifact_id or f"artifact-{uuid.uuid4().hex[:12]}"
    record = ArtifactRecord(
        artifact_id=selected_id,
        run_id=run.run_id,
        artifact_type=artifact_type,
        mime_type=mime_type,
        relative_path=relative,
        sha256=digest,
        size_bytes=size,
        producer_task_id=producer_task_id,
        producer_tool=producer_tool,
        provenance=dict(provenance or {}),
        trust=ArtifactTrust(trust or ArtifactTrust.INTERNAL),
    )
    if record.artifact_id in run.artifacts:
        raise ValueError(f"duplicate artifact_id: {record.artifact_id}")
    existing_input_id = run.workflow_inputs.get(record.artifact_type)
    if existing_input_id is not None and existing_input_id != record.artifact_id:
        raise ValueError(
            f"workflow input contract {record.artifact_type!r} is already bound "
            f"to artifact {existing_input_id!r}"
        )
    if producer_task_id:
        task_contract = _run_task_contract(run, producer_task_id)
        if task_contract is not None and artifact_type not in set(
            task_contract.get("output_artifacts", ())
        ):
            raise ValueError(
                f"task {producer_task_id!r} is not contracted to produce "
                f"artifact type {artifact_type!r}"
            )

    def authorize_registration(
        authoritative_run: WorkflowRun | None,
        events: Sequence[WorkflowEvent],
    ) -> None:
        _authorize_artifact_registration(
            authoritative_run,
            events,
            record=record,
            active_task_id=active_task_id,
        )

    self.append_event(
        "artifact_registered",
        {"artifact": record.to_dict()},
        precondition=authorize_registration,
    )
    return self._require_run().artifacts[record.artifact_id]

verify_task_inputs(task_id)

Verify the selected input artifacts for one task before execution.

Source code in src/cs_copilot/workflows/runtime.py

def verify_task_inputs(self, task_id: str) -> list[ArtifactRecord]:
    """Verify the selected input artifacts for one task before execution."""

    selected_task_id = validate_identifier(task_id, field="task_id")
    run = self._require_run()
    try:
        task = run.tasks[selected_task_id]
    except KeyError as exc:
        raise ValueError(f"unknown task_id: {selected_task_id}") from exc
    input_gaps = _task_input_gaps(run, selected_task_id)
    if input_gaps:
        raise InvalidTransitionError(
            f"task {selected_task_id!r} has invalid required input "
            f"artifacts: {'; '.join(input_gaps)}"
        )
    return [
        self.verify_artifact(artifact_id)
        for artifact_id in dict.fromkeys(task.input_artifact_ids)
    ]

rebuild_snapshots()

Replay immutable events and replace both derived snapshots.

Source code in src/cs_copilot/workflows/runtime.py

def rebuild_snapshots(self) -> WorkflowRun:
    """Replay immutable events and replace both derived snapshots."""

    events = _read_events(self.layout)
    run = _replay(events)
    self.events = events
    self.run = run
    self._syncs_since_full = 0
    self.layout = OutputLayout(run.session_id, run.run_id, run.workflow_slug)
    self._write_snapshots()
    return run

complete(*, required_artifact_types=None, required_task_ids=None)

Complete the run only after its catalog contracts are satisfied.

Catalog requirements are authoritative and explicit requirements can only make them stricter. Ad-hoc runs require every registered task and any explicitly requested artifacts to be complete.

Source code in src/cs_copilot/workflows/runtime.py

def complete(
    self,
    *,
    required_artifact_types: Sequence[str] | None = None,
    required_task_ids: Sequence[str] | None = None,
) -> WorkflowRun:
    """Complete the run only after its catalog contracts are satisfied.

    Catalog requirements are authoritative and explicit requirements can
    only make them stricter. Ad-hoc runs require every registered task and
    any explicitly requested artifacts to be complete.
    """

    run = self._require_run()
    if run.status in {RunStatus.COMPLETED, RunStatus.PARTIAL}:
        return run
    if run.status in {RunStatus.FAILED, RunStatus.CANCELLED}:
        raise InvalidTransitionError(
            f"cannot complete terminal run in status {run.status.value!r}"
        )
    if run.status is not RunStatus.RUNNING:
        run = self.transition_run(
            RunStatus.RUNNING,
            reason="completion validation requested",
        )
    self.verify_artifacts()
    reasons = _completion_gaps(
        run,
        required_artifact_types=required_artifact_types,
        required_task_ids=required_task_ids,
    )
    if reasons:
        return self.transition_run(RunStatus.PARTIAL, reason="; ".join(reasons))
    return self.transition_run(RunStatus.COMPLETED)

RunStatus

Bases: str, Enum

Lifecycle states for a workflow run.

Source code in src/cs_copilot/workflows/runtime.py

class RunStatus(str, Enum):
    """Lifecycle states for a workflow run."""

    SUBMITTED = "submitted"
    PLANNING = "planning"
    RUNNING = "running"
    INPUT_REQUIRED = "input_required"
    COMPLETED = "completed"
    PARTIAL = "partial"
    FAILED = "failed"
    CANCELLED = "cancelled"

RunStore

Read-only adapter used by trajectory replay and external evaluators.

Source code in src/cs_copilot/workflows/runtime.py

class RunStore:
    """Read-only adapter used by trajectory replay and external evaluators."""

    def __init__(self, *, session_id: str | None = None) -> None:
        self.session_id = session_id

    def events(self, run_id: str) -> list[dict[str, Any]]:
        context = RunContext.load(run_id, session_id=self.session_id)
        return [event.to_dict() for event in context.events]

    def rebuild(self, run_id: str) -> WorkflowRun:
        context = RunContext.load(run_id, session_id=self.session_id)
        return context.rebuild_snapshots()

TaskRecord dataclass

Durable state for one role-assigned workflow task.

Source code in src/cs_copilot/workflows/runtime.py

@dataclass
class TaskRecord:
    """Durable state for one role-assigned workflow task."""

    task_id: str
    role: str
    profile: str
    step: str
    status: TaskStatus = TaskStatus.PENDING
    attempts: int = 0
    input_artifact_ids: list[str] = field(default_factory=list)
    output_artifact_ids: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=lambda: _utc_now())
    updated_at: str = field(default_factory=lambda: _utc_now())
    error: ToolError | None = None

    def __post_init__(self) -> None:
        self.task_id = validate_identifier(self.task_id, field="task_id")
        self.status = TaskStatus(self.status)
        if self.attempts < 0:
            raise ValueError("task attempts cannot be negative")
        if not self.role.strip() or not self.profile.strip() or not self.step.strip():
            raise ValueError("task role, profile, and step are required")

    def to_dict(self) -> dict[str, Any]:
        return {
            "task_id": self.task_id,
            "role": self.role,
            "profile": self.profile,
            "step": self.step,
            "status": self.status.value,
            "attempts": self.attempts,
            "input_artifact_ids": list(self.input_artifact_ids),
            "output_artifact_ids": list(self.output_artifact_ids),
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "error": self.error.to_dict() if self.error else None,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "TaskRecord":
        error = value.get("error")
        return cls(
            task_id=str(value["task_id"]),
            role=str(value["role"]),
            profile=str(value["profile"]),
            step=str(value["step"]),
            status=TaskStatus(str(value.get("status", TaskStatus.PENDING.value))),
            attempts=int(value.get("attempts", 0)),
            input_artifact_ids=[str(item) for item in value.get("input_artifact_ids", ())],
            output_artifact_ids=[str(item) for item in value.get("output_artifact_ids", ())],
            created_at=str(value.get("created_at") or _utc_now()),
            updated_at=str(value.get("updated_at") or _utc_now()),
            error=ToolError.from_dict(error) if isinstance(error, Mapping) else None,
        )

TaskStatus

Bases: str, Enum

Lifecycle states for one workflow task.

Source code in src/cs_copilot/workflows/runtime.py

class TaskStatus(str, Enum):
    """Lifecycle states for one workflow task."""

    PENDING = "pending"
    RUNNING = "running"
    INPUT_REQUIRED = "input_required"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    SKIPPED = "skipped"

ToolError dataclass

Normalized tool/workflow error suitable for persistence and retry policy.

Source code in src/cs_copilot/workflows/runtime.py

@dataclass(frozen=True)
class ToolError:
    """Normalized tool/workflow error suitable for persistence and retry policy."""

    code: ToolErrorCode
    message: str
    retryable: bool = False
    details: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "code", ToolErrorCode(self.code))
        if not str(self.message).strip():
            raise ValueError("ToolError.message cannot be empty")

    def to_dict(self) -> dict[str, Any]:
        return {
            "code": self.code.value,
            "message": self.message,
            "retryable": self.retryable,
            "details": _json_safe(self.details),
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "ToolError":
        return cls(
            code=ToolErrorCode(str(value["code"])),
            message=str(value["message"]),
            retryable=bool(value.get("retryable", False)),
            details=dict(value.get("details") or {}),
        )

ToolErrorCode

Bases: str, Enum

Stable error taxonomy shared by workflow and tool envelopes.

Source code in src/cs_copilot/execution/errors.py

class ToolErrorCode(str, Enum):
    """Stable error taxonomy shared by workflow and tool envelopes."""

    INVALID_INPUT = "invalid_input"
    PERMISSION_DENIED = "permission_denied"
    TRANSIENT_EXTERNAL = "transient_external"
    TIMEOUT = "timeout"
    RESOURCE_LIMIT = "resource_limit"
    SCIENTIFIC_VALIDATION = "scientific_validation"
    INTERNAL = "internal"

WorkflowEvent dataclass

One immutable workflow event object.

Source code in src/cs_copilot/workflows/runtime.py

@dataclass(frozen=True)
class WorkflowEvent:
    """One immutable workflow event object."""

    event_id: str
    sequence: int
    run_id: str
    event_type: str
    timestamp: str
    payload: Mapping[str, Any]
    schema_version: int = SCHEMA_VERSION

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_version": self.schema_version,
            "event_id": self.event_id,
            "sequence": self.sequence,
            "run_id": self.run_id,
            "event_type": self.event_type,
            "timestamp": self.timestamp,
            "payload": _json_safe(self.payload),
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "WorkflowEvent":
        schema_version = int(value.get("schema_version", 0))
        if schema_version != SCHEMA_VERSION:
            raise EventReplayError(
                f"unsupported event schema version {schema_version}; expected {SCHEMA_VERSION}"
            )
        return cls(
            schema_version=schema_version,
            event_id=validate_identifier(value["event_id"], field="event_id"),
            sequence=int(value["sequence"]),
            run_id=validate_identifier(value["run_id"], field="run_id"),
            event_type=str(value["event_type"]),
            timestamp=str(value["timestamp"]),
            payload=dict(value.get("payload") or {}),
        )

WorkflowRun dataclass

Current state reconstructed from a workflow run's event stream.

Source code in src/cs_copilot/workflows/runtime.py

@dataclass
class WorkflowRun:
    """Current state reconstructed from a workflow run's event stream."""

    session_id: str
    run_id: str
    workflow_slug: str
    trace_id: str
    workflow_contract: dict[str, Any] = field(default_factory=dict)
    status: RunStatus = RunStatus.SUBMITTED
    constraints: dict[str, Any] = field(default_factory=dict)
    budget: dict[str, Any] = field(default_factory=dict)
    workflow_inputs: dict[str, str] = field(default_factory=dict)
    tasks: dict[str, TaskRecord] = field(default_factory=dict)
    artifacts: dict[str, ArtifactRecord] = field(default_factory=dict)
    handoffs: list[HandoffEnvelope] = field(default_factory=list)
    created_at: str = field(default_factory=lambda: _utc_now())
    updated_at: str = field(default_factory=lambda: _utc_now())
    error: ToolError | None = None
    schema_version: int = SCHEMA_VERSION

    def __post_init__(self) -> None:
        self.session_id = validate_identifier(self.session_id, field="session_id")
        self.run_id = validate_identifier(self.run_id, field="run_id")
        self.workflow_slug = sanitize_workflow_slug(self.workflow_slug)
        self.status = RunStatus(self.status)
        self.workflow_contract = dict(self.workflow_contract or {})
        if self.schema_version != SCHEMA_VERSION:
            raise ValueError(
                f"unsupported workflow schema version {self.schema_version}; expected {SCHEMA_VERSION}"
            )
        if not self.trace_id.strip():
            raise ValueError("trace_id cannot be empty")
        if self.workflow_contract:
            _validate_workflow_contract_snapshot(
                self.workflow_contract,
                expected_slug=self.workflow_slug,
            )

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_version": self.schema_version,
            "session_id": self.session_id,
            "run_id": self.run_id,
            "workflow_slug": self.workflow_slug,
            "trace_id": self.trace_id,
            "workflow_contract": _json_safe(self.workflow_contract),
            "status": self.status.value,
            "constraints": _json_safe(self.constraints),
            "budget": _json_safe(self.budget),
            "workflow_inputs": dict(sorted(self.workflow_inputs.items())),
            "tasks": [self.tasks[key].to_dict() for key in sorted(self.tasks)],
            "artifacts": [self.artifacts[key].to_dict() for key in sorted(self.artifacts)],
            "handoffs": [handoff.to_dict() for handoff in self.handoffs],
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "error": self.error.to_dict() if self.error else None,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "WorkflowRun":
        tasks = [TaskRecord.from_dict(item) for item in value.get("tasks", ())]
        artifacts = [ArtifactRecord.from_dict(item) for item in value.get("artifacts", ())]
        error = value.get("error")
        return cls(
            schema_version=int(value.get("schema_version", 0)),
            session_id=str(value["session_id"]),
            run_id=str(value["run_id"]),
            workflow_slug=str(value["workflow_slug"]),
            trace_id=str(value["trace_id"]),
            workflow_contract=dict(value.get("workflow_contract") or {}),
            status=RunStatus(str(value.get("status", RunStatus.SUBMITTED.value))),
            constraints=dict(value.get("constraints") or {}),
            budget=dict(value.get("budget") or {}),
            workflow_inputs={
                str(name): validate_identifier(artifact_id, field="workflow input artifact_id")
                for name, artifact_id in dict(value.get("workflow_inputs") or {}).items()
            },
            tasks={task.task_id: task for task in tasks},
            artifacts={artifact.artifact_id: artifact for artifact in artifacts},
            handoffs=[HandoffEnvelope.from_dict(item) for item in value.get("handoffs", ())],
            created_at=str(value.get("created_at") or _utc_now()),
            updated_at=str(value.get("updated_at") or _utc_now()),
            error=ToolError.from_dict(error) if isinstance(error, Mapping) else None,
        )

WorkflowRuntimeError

Bases: RuntimeError

Base error for workflow runtime contract failures.

Source code in src/cs_copilot/workflows/runtime.py

class WorkflowRuntimeError(RuntimeError):
    """Base error for workflow runtime contract failures."""

prepare_chembl_retrieval(target=None, target_type=None, organism=None, assay_types=None, mechanism=None, notes=None)

Validate a prospective ChEMBL retrieval from caller-supplied fields.

The connected LLM extracts these from the user request; this gate only checks completeness and returns canonical clarifying questions for any missing dimension.

Parameters

target: The specific target the user confirmed — a gene symbol (e.g. EPHX2), protein name (e.g. soluble epoxide hydrolase), ChEMBL target id, or an organism-level target. Leave empty/None if the user has not named a specific target yet; do not infer one. target_type: "protein" (default when a target is given) or "organism". Organism targets do not require a separate organism filter. organism: Organism filter (e.g. Homo sapiens, Mus musculus, all species). assay_types: Assay types to include — ChEMBL codes (B/F/A/...) or words (binding/functional/admet). mechanism: A specific mechanism of action, or "any"/"unspecified" for no mechanism filter. notes: Optional free-text context. Recorded for the caller; never parsed for the decision.

Source code in src/cs_copilot/workflows/chembl_policy.py

def prepare_chembl_retrieval(
    target: str | None = None,
    target_type: str | None = None,
    organism: str | None = None,
    assay_types: list[str] | str | None = None,
    mechanism: str | None = None,
    notes: str | None = None,
) -> dict[str, Any]:
    """Validate a prospective ChEMBL retrieval from caller-supplied fields.

    The connected LLM extracts these from the user request; this gate only
    checks completeness and returns canonical clarifying questions for any
    missing dimension.

    Parameters
    ----------
    target:
        The specific target the user confirmed — a gene symbol (e.g. ``EPHX2``),
        protein name (e.g. ``soluble epoxide hydrolase``), ChEMBL target id, or an
        organism-level target. Leave empty/None if the user has not named a
        specific target yet; do not infer one.
    target_type:
        ``"protein"`` (default when a target is given) or ``"organism"``. Organism
        targets do not require a separate organism filter.
    organism:
        Organism filter (e.g. ``Homo sapiens``, ``Mus musculus``, ``all species``).
    assay_types:
        Assay types to include — ChEMBL codes (``B``/``F``/``A``/...) or words
        (``binding``/``functional``/``admet``).
    mechanism:
        A specific mechanism of action, or ``"any"``/``"unspecified"`` for no
        mechanism filter.
    notes:
        Optional free-text context. Recorded for the caller; never parsed for the
        decision.
    """

    target_clean = _clean(target) or None
    resolved_type = _normalize_target_type(target_type, target_clean)
    organism_clean = _clean(organism) or None
    codes = _normalize_assay_types(assay_types)
    mechanism_value, mechanism_preference = _normalize_mechanism(mechanism)

    missing: list[str] = []
    questions: list[str] = []

    if not target_clean:
        missing.append("target_specificity")
        questions.append(
            "Which specific target should ChEMBL search — a gene symbol, protein name, "
            "ChEMBL target id, or organism-level target?"
        )

    if resolved_type != "organism" and not organism_clean:
        missing.append("organism")
        questions.append(
            "Which organism should be used for the ChEMBL target or assay filter "
            "(for example, Homo sapiens, Mus musculus, or all species)?"
        )

    if not codes:
        missing.append("assay_types")
        questions.append(
            "Which assay types should be included: binding, functional, ADMET, or a combination?"
        )

    if mechanism_preference is None:
        missing.append("mechanism")
        questions.append(
            "Should assays be filtered to a specific mechanism of action, or should mechanism "
            "be unspecified/any?"
        )

    can_proceed = not missing
    return ChemblRetrievalDecision(
        can_proceed=can_proceed,
        needs_clarification=not can_proceed,
        missing_requirements=missing,
        clarifying_questions=questions,
        target=target_clean,
        target_type=resolved_type,
        organism=organism_clean,
        assay_types=codes,
        mechanism=mechanism_value,
        mechanism_preference=mechanism_preference,
        recommended_next_tool="chembl_convert_to_chembl_query" if can_proceed else None,
    ).as_dict()

plan_chemical_space_analysis(analysis_intents=None, dataset_source=None, notes=None)

Validate a chemical-space analysis plan from caller-supplied fields.

Parameters

analysis_intents: One or more of: chembl_retrieval, gtm_build, gtm_analysis, density_landscape, activity_landscape, gtm_projection, chemotype_sar_analysis, report_generation. Leave empty if the user has not narrowed a broad "analyze chemical space" request; do not infer. dataset_source: One of session_clean_dataset, explicit_path, uploaded_dataset, or chembl_retrieval. notes: Optional free-text context; recorded, never parsed for the decision.

Source code in src/cs_copilot/workflows/chemical_space_policy.py

def plan_chemical_space_analysis(
    analysis_intents: list[str] | str | None = None,
    dataset_source: str | None = None,
    notes: str | None = None,
) -> dict[str, Any]:
    """Validate a chemical-space analysis plan from caller-supplied fields.

    Parameters
    ----------
    analysis_intents:
        One or more of: ``chembl_retrieval``, ``gtm_build``, ``gtm_analysis``,
        ``density_landscape``, ``activity_landscape``, ``gtm_projection``,
        ``chemotype_sar_analysis``, ``report_generation``. Leave empty if the user
        has not narrowed a broad "analyze chemical space" request; do not infer.
    dataset_source:
        One of ``session_clean_dataset``, ``explicit_path``, ``uploaded_dataset``,
        or ``chembl_retrieval``.
    notes:
        Optional free-text context; recorded, never parsed for the decision.
    """

    intents = _normalize_intents(analysis_intents)
    source = _clean(dataset_source) or None
    missing: list[str] = []
    questions: list[str] = []
    plan_notes: list[str] = []

    if not intents:
        missing.append("analysis_intent")
        questions.append(
            "Which analysis do you want: ChEMBL retrieval, GTM density map, activity "
            "landscape, projection, chemotype/SAR analysis, or report generation?"
        )

    if "chembl_retrieval" in intents:
        plan_notes.append("Use chembl_prepare_retrieval before calling ChEMBL retrieval tools.")
        source = source or "chembl_retrieval"

    if not source:
        missing.append("data_source")
        questions.append(
            "Which data source should be used: an existing session clean dataset, an explicit "
            "CSV/Parquet path, uploaded data, or a new ChEMBL retrieval?"
        )

    can_proceed = not missing
    return ChemicalSpaceDecision(
        can_proceed=can_proceed,
        needs_clarification=not can_proceed,
        missing_requirements=missing,
        clarifying_questions=questions,
        analysis_intents=intents,
        dataset_source=source,
        recommended_next_tools=_recommended_tools(intents, can_proceed),
        notes=plan_notes,
    ).as_dict()