spakky-agent

spakky-agent는 Agent workflow를 Spakky 컴포넌트로 모델링하기 위한 계약, 도구, 상태, signal, evidence 타입을 제공합니다.

Agentic Hexagonal Architecture의 core 계약입니다.

설치

pip install spakky-agent

spakky-agent는 @Agent, AgentExecutionSpec, RunAgentInput, AgentRunner, AgentEvent, AgentYield, tool dispatch, context compaction, state/signal/evidence repository port, task store, safety/recovery/delegation 타입 같은 public contract를 소유합니다. 이 패키지는 의도적으로 vLLM, SQLAlchemy, FastAPI, Typer, AG-UI, A2A, MCP를 import하지 않습니다. 운영에서 durable execution을 사용하려면 provider contribution의 repository 구현이 필요하며, 운영용 in-memory fallback은 제공하지 않습니다.

Public API

Agentic hexagonal architecture contracts for Spakky Framework.

Message = Progress module-attribute

Backward-compatible progress-yield payload alias.

TextDelta = Token module-attribute

Backward-compatible token-yield payload alias.

PLUGIN_NAME = Plugin(name='spakky-agent') module-attribute

Plugin identifier for the Spakky Agent package.

AgentCancellationCleanupCallable = Callable[[AgentCancellationRequest], Awaitable[AgentCancellationCleanupResult]]

Async hook shape invoked for model stream/tool/delegate cancellation cleanup.

AgentEvent = MessageDeltaEvent | ReasoningDeltaEvent | ToolCallStartEvent | ToolCallArgsDeltaEvent | ToolCallEndEvent | ToolCallResultEvent | RunStartedEvent | RunPausedEvent | RunFinishedEvent | StepStartedEvent | StepFinishedEvent | StateSnapshotEvent | StateDeltaEvent | ArtifactEvent

Discriminated union of every protocol-neutral agent event.

Each member exposes a distinct kind literal so adapters dispatch with match/case exhaustiveness rather than runtime attribute probing.

JsonObject = Mapping[str, JsonValue]

JSON object payload used by public agent contracts.

JsonPrimitive = bool | float | int | str | None

Scalar JSON value accepted by agent public contracts.

JsonValue = JsonPrimitive | Mapping[str, JsonValue] | Sequence[JsonValue]

Recursive JSON-compatible value used at model, signal, and evidence boundaries.

AbstractSpakkyAgentError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for agent-related errors.

AgentBootstrapError

Bases: AbstractSpakkyAgentError

Raised when agent bootstrap validation fails.

AgentDefinitionError

Bases: AbstractSpakkyAgentError

Raised when an agent contract cannot be defined safely.

AgentModelConfigurationError

Bases: AgentBootstrapError

Raised when an agent requires a model adapter but none is registered.

AgentOutputGuardError

Bases: AbstractSpakkyAgentError

Raised when an output guard detects unsafe streaming exposure.

AgentPersistenceConfigurationError

Bases: AgentBootstrapError

Raised when production agent persistence is required but not provided.

AgentToolBindingError

Bases: AbstractSpakkyAgentError

Raised when a model tool-call payload cannot be bound safely.

AgentToolDispatchError

Bases: AbstractSpakkyAgentError

Raised when a model tool call names a tool absent from the catalog.

AgentToolDispatcher(target, catalog, runtime_context=None) dataclass

Resolve and invoke a catalog tool from a model tool call.

The dispatcher binds to a single agent instance whose @agent_tool methods are described by catalog. Catalog descriptors that own no instance parameter (such as MCP-normalized external tools) are invoked without the bound target.

descriptor_for(call)

Resolve the catalog descriptor a model tool call targets.

Source code in core/spakky-agent/src/spakky/agent/dispatcher.py

def descriptor_for(self, call: ModelToolCall) -> AgentToolDescriptor:
    """Resolve the catalog descriptor a model tool call targets."""
    for descriptor in self.catalog.descriptors:
        if descriptor.schema.name == call.name:
            return descriptor
    raise AgentToolDispatchError("Agent tool call names an unregistered tool")

dispatch(call) async

Bind a model tool call payload and invoke its catalog callable.

Source code in core/spakky-agent/src/spakky/agent/dispatcher.py

async def dispatch(self, call: ModelToolCall) -> object:
    """Bind a model tool call payload and invoke its catalog callable."""
    descriptor = self.descriptor_for(call)
    invocation = descriptor.bind_invocation(call.arguments)
    callable_ = descriptor.callable
    positional = self._with_injected_prefix(callable_, invocation.args)
    if iscoroutinefunction(callable_):
        awaitable: Awaitable[object] = callable_(
            *positional,
            **invocation.kwargs,
        )
        return await awaitable
    return callable_(*positional, **invocation.kwargs)

ContextDigest(id, context_identity, source_manifest_ref, digest, derived_from_pack_ids=tuple(), compression_evidence_ref=None, algorithm=None, summary=None, created_at=None, metadata=dict()) dataclass

Derived compression evidence for a context identity.

ContextFreshness

Bases: StrEnum

Freshness classification for context rot and budget decisions.

ContextHealthSignal(id, symptom, manifest_ref=None, pack_id=None, evidence_ref=None, score=None, observed_at=None, metadata=dict()) dataclass

Observed context rot signal used to choose optimization actions.

evidence_payload()

Return JSON-compatible signal metadata for append-only evidence.

Source code in core/spakky-agent/src/spakky/agent/context.py

def evidence_payload(self) -> Mapping[str, JsonValue]:
    """Return JSON-compatible signal metadata for append-only evidence."""
    return {
        "id": self.id,
        "symptom": self.symptom.value,
        "manifest_ref": self.manifest_ref,
        "pack_id": self.pack_id,
        "evidence_ref": self.evidence_ref,
        "score": self.score,
        "observed_at": self.observed_at.isoformat()
        if self.observed_at is not None
        else None,
        "metadata": self.metadata,
    }

ContextManifest(id, entries, origin_ref=None, evidence_refs=tuple(), created_at=None, metadata=dict()) dataclass

Auditable composition record for model input context packs.

ContextManifestEntry(pack_id, source, role, origin_ref, evidence_ref=None, digest_ref=None, sensitive_fields=(), metadata=dict()) dataclass

One audited pack entry inside a context manifest.

ContextOptimizationAction(id, kind, signal_refs=tuple(), target_pack_ids=tuple(), manifest_ref=None, digest_ref=None, delegation_ref=None, result_evidence_ref=None, reason=None, metadata=dict()) dataclass

Selected optimization action derived from context health signals.

evidence_payload()

Return JSON-compatible action metadata without raw context contents.

Source code in core/spakky-agent/src/spakky/agent/context.py

def evidence_payload(self) -> Mapping[str, JsonValue]:
    """Return JSON-compatible action metadata without raw context contents."""
    return {
        "id": self.id,
        "kind": self.kind.value,
        "signal_refs": tuple(self.signal_refs),
        "target_pack_ids": tuple(self.target_pack_ids),
        "manifest_ref": self.manifest_ref,
        "digest_ref": self.digest_ref,
        "delegation_ref": self.delegation_ref,
        "result_evidence_ref": self.result_evidence_ref,
        "reason": self.reason,
        "metadata": self.metadata,
    }

ContextOptimizationActionKind

Bases: StrEnum

Optimization actions that can be selected from context health signals.

ContextOptimizationEvidenceStage

Bases: StrEnum

Where an optimization action evidence item sits in the agent flow.

ContextPack(id, content, source, role, freshness=ContextFreshness.UNKNOWN, relevance=None, token_budget=ContextTokenBudget(), sensitivity=ContextSensitivity.INTERNAL, sensitive_fields=(), metadata=dict()) dataclass

LLM-facing context unit derived from state, signal, or evidence.

guarded_content(policy=None)

Return deterministic model-safe content for this context pack.

Source code in core/spakky-agent/src/spakky/agent/context.py

def guarded_content(
    self,
    policy: ContextExposurePolicy | None = None,
) -> str:
    """Return deterministic model-safe content for this context pack."""
    exposure_policy = policy or ContextExposurePolicy()
    if self.sensitivity == ContextSensitivity.REDACTED:
        return "[REDACTED]"
    guarded = guard_json_value(
        {"content": self.content},
        tuple(
            SensitiveFieldDescriptor(
                ("content", *descriptor.path), descriptor.field
            )
            for descriptor in self.sensitive_fields
        ),
        exposure_policy,
    )
    content = cast(Mapping[str, JsonValue], guarded).get("content")
    if isinstance(content, str):
        return content
    return "[REDACTED]"

message_metadata(policy=None)

Return non-content metadata for provider-neutral model messages.

Source code in core/spakky-agent/src/spakky/agent/context.py

def message_metadata(
    self,
    policy: ContextExposurePolicy | None = None,
) -> Mapping[str, JsonValue]:
    """Return non-content metadata for provider-neutral model messages."""
    exposure_policy = policy or ContextExposurePolicy()
    metadata: dict[str, JsonValue] = {
        "context_pack_id": self.id,
        "source": self.source,
        "role": self.role.value,
        "freshness": self.freshness.value,
        "relevance": self.relevance,
        "token_budget": {
            "max_tokens": self.token_budget.max_tokens,
            "estimated_tokens": self.token_budget.estimated_tokens,
            "reserved_output_tokens": self.token_budget.reserved_output_tokens,
        },
        "sensitivity": self.sensitivity.value,
        "metadata": self.metadata,
    }
    if exposure_policy.include_sensitive_context_metadata and self.sensitive_fields:
        metadata["sensitive_fields"] = tuple(
            descriptor.to_metadata() for descriptor in self.sensitive_fields
        )
    return metadata

ContextPackRole

Bases: StrEnum

Semantic role of a context pack inside a model request.

ContextRotSymptom

Bases: StrEnum

Typed context rot symptoms observed before model input assembly.

ContextSensitivity

Bases: StrEnum

Deterministic sensitivity metadata carried before model input.

ContextTokenBudget(max_tokens=None, estimated_tokens=None, reserved_output_tokens=None) dataclass

Token budget allocated to one context pack.

IAgentContextHandler

Bases: ABC

Select context optimization actions from health signals and manifests.

select_optimization_actions(signals, manifest) abstractmethod

Return optimization actions without mutating raw evidence.

Source code in core/spakky-agent/src/spakky/agent/context.py

@abstractmethod
def select_optimization_actions(
    self,
    signals: Sequence[ContextHealthSignal],
    manifest: "ContextManifest",
) -> Sequence[ContextOptimizationAction]:
    """Return optimization actions without mutating raw evidence."""
    ...

AgentCancellationCleanupReport(state_id, signal_id, outcomes=()) dataclass

Aggregate cleanup evidence for one CANCEL signal.

cleanup_succeeded property

Return whether all cleanup hooks completed without failure.

failed_outcomes property

Return the hook outcomes that force FAILED terminal state.

__post_init__()

Reject reports that cannot be attached to state and signal evidence.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject reports that cannot be attached to state and signal evidence."""
    _require_non_blank(self.state_id, "Agent cancellation report state id")
    _require_non_blank(self.signal_id, "Agent cancellation report signal id")

to_payload()

Serialize the report for state metadata and evidence payloads.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_payload(self) -> JsonObject:
    """Serialize the report for state metadata and evidence payloads."""
    return {
        "state_id": self.state_id,
        "signal_id": self.signal_id,
        "cleanup_succeeded": self.cleanup_succeeded,
        "outcomes": tuple(outcome.to_payload() for outcome in self.outcomes),
    }

to_evidence_candidate(*, summary=None)

Represent cancellation cleanup as append-only evidence.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_evidence_candidate(
    self,
    *,
    summary: str | None = None,
) -> AgentEvidenceCandidate:
    """Represent cancellation cleanup as append-only evidence."""
    return AgentEvidenceCandidate(
        kind=AgentEvidenceKind.CANCELLATION,
        payload=self.to_payload(),
        summary=summary,
    )

AgentCancellationCleanupResult(target_kind, target_ref, status, reason=None, error_code=None, message=None, metadata=dict()) dataclass

Result returned by one cancellation cleanup hook.

failed_cleanup property

Return whether this hook prevents a clean CANCELLED terminal state.

__post_init__()

Reject outcomes that cannot be correlated with the cleanup target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject outcomes that cannot be correlated with the cleanup target."""
    _require_non_blank(self.target_ref, "Agent cancellation cleanup target ref")
    _require_optional_non_blank(self.reason, "Agent cancellation cleanup reason")
    _require_optional_non_blank(
        self.error_code,
        "Agent cancellation cleanup error code",
    )
    _require_optional_non_blank(self.message, "Agent cancellation cleanup message")

succeeded(*, target_kind, target_ref, reason=None, metadata=None) classmethod

Record a cleanup hook that released its target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def succeeded(
    cls,
    *,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    reason: str | None = None,
    metadata: JsonObject | None = None,
) -> "AgentCancellationCleanupResult":
    """Record a cleanup hook that released its target."""
    return cls(
        target_kind=target_kind,
        target_ref=target_ref,
        status=AgentCancellationCleanupStatus.SUCCEEDED,
        reason=reason,
        metadata={} if metadata is None else metadata,
    )

failed(*, target_kind, target_ref, error_code, message, metadata=None) classmethod

Record a cleanup hook that could not release its target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def failed(
    cls,
    *,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    error_code: str,
    message: str,
    metadata: JsonObject | None = None,
) -> "AgentCancellationCleanupResult":
    """Record a cleanup hook that could not release its target."""
    return cls(
        target_kind=target_kind,
        target_ref=target_ref,
        status=AgentCancellationCleanupStatus.FAILED,
        error_code=error_code,
        message=message,
        metadata={} if metadata is None else metadata,
    )

skipped(*, target_kind, target_ref, reason, metadata=None) classmethod

Record a cleanup target that was already inactive or unavailable.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def skipped(
    cls,
    *,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    reason: str,
    metadata: JsonObject | None = None,
) -> "AgentCancellationCleanupResult":
    """Record a cleanup target that was already inactive or unavailable."""
    return cls(
        target_kind=target_kind,
        target_ref=target_ref,
        status=AgentCancellationCleanupStatus.SKIPPED,
        reason=reason,
        metadata={} if metadata is None else metadata,
    )

to_payload()

Serialize this hook outcome into evidence/state metadata.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_payload(self) -> JsonObject:
    """Serialize this hook outcome into evidence/state metadata."""
    return {
        "target_kind": self.target_kind.value,
        "target_ref": self.target_ref,
        "status": self.status.value,
        "reason": self.reason,
        "error_code": self.error_code,
        "message": self.message,
        "metadata": self.metadata,
    }

AgentCancellationCleanupStatus

Bases: StrEnum

Outcome of one cancellation cleanup hook.

AgentCancellationCleanupTask(target_kind, target_ref, cleanup, metadata=dict()) dataclass

One cleanup hook registered for a running cancellation target.

__post_init__()

Reject hooks that cannot be identified in cleanup evidence.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject hooks that cannot be identified in cleanup evidence."""
    _require_non_blank(self.target_ref, "Agent cancellation cleanup target ref")

run(*, state, signal) async

Invoke the hook and verify that it reports the same target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

async def run(
    self,
    *,
    state: AgentState,
    signal: AgentSignal,
) -> AgentCancellationCleanupResult:
    """Invoke the hook and verify that it reports the same target."""
    request = AgentCancellationRequest.from_signal(
        state=state,
        signal=signal,
        target_kind=self.target_kind,
        target_ref=self.target_ref,
        metadata=self.metadata,
    )
    result = await self.cleanup(request)
    if (
        result.target_kind is not self.target_kind
        or result.target_ref != self.target_ref
    ):
        raise AgentDefinitionError(
            "Agent cancellation cleanup result target must match the task"
        )
    return result

AgentCancellationRequest(state_id, signal_id, target_kind, target_ref, reason=None, requested_by=None, metadata=dict()) dataclass

Cancellation request passed to a model stream, tool, or delegate hook.

__post_init__()

Reject requests that cannot be traced back to state/signal/target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject requests that cannot be traced back to state/signal/target."""
    _require_non_blank(self.state_id, "Agent cancellation state id")
    _require_non_blank(self.signal_id, "Agent cancellation signal id")
    _require_non_blank(self.target_ref, "Agent cancellation target ref")
    _require_optional_non_blank(self.reason, "Agent cancellation reason")
    _require_optional_non_blank(self.requested_by, "Agent cancellation requester")

from_signal(*, state, signal, target_kind, target_ref, metadata=None) classmethod

Build a hook request from the durable CANCEL signal payload.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def from_signal(
    cls,
    *,
    state: AgentState,
    signal: AgentSignal,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    metadata: JsonObject | None = None,
) -> "AgentCancellationRequest":
    """Build a hook request from the durable CANCEL signal payload."""
    _require_cancel_signal(signal)
    _require_signal_matches_state(state, signal)
    return cls(
        state_id=state.id,
        signal_id=signal.id,
        target_kind=target_kind,
        target_ref=target_ref,
        reason=_optional_string(signal.payload, "reason"),
        requested_by=_optional_string(signal.payload, "requested_by"),
        metadata={} if metadata is None else metadata,
    )

to_payload()

Serialize the request into append-only evidence metadata.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_payload(self) -> JsonObject:
    """Serialize the request into append-only evidence metadata."""
    return {
        "state_id": self.state_id,
        "signal_id": self.signal_id,
        "target_kind": self.target_kind.value,
        "target_ref": self.target_ref,
        "reason": self.reason,
        "requested_by": self.requested_by,
        "metadata": self.metadata,
    }

AgentCancellationTargetKind

Bases: StrEnum

Running execution target that can receive a cancellation cleanup hook.

AgentEvidence(id, agent_state_id, kind, payload=dict(), summary=None, digest=None, manifest_ref=None, reference=None, sensitive_fields=(), created_at=None) dataclass

Append-only artifact captured during an agent execution.

AgentEvidenceCandidate(kind, payload=dict(), summary=None, digest=None, manifest_ref=None, reference=None, sensitive_fields=()) dataclass

Append-only evidence candidate before repository id assignment.

tool_result(*, tool_identity, tool_schema_name, result, capture, sensitive_fields=(), exposure_policy=None, summary=None) classmethod

Create evidence metadata for a captured tool result.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def tool_result(
    cls,
    *,
    tool_identity: str,
    tool_schema_name: str,
    result: JsonObject,
    capture: EvidenceCapture,
    sensitive_fields: tuple[SensitiveFieldDescriptor, ...] = (),
    exposure_policy: EvidenceExposurePolicy | None = None,
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create evidence metadata for a captured tool result."""
    _require_non_blank(tool_identity, "Agent tool identity")
    _require_non_blank(tool_schema_name, "Agent tool schema name")
    policy = exposure_policy or EvidenceExposurePolicy()
    guarded_result = guard_json_value(result, sensitive_fields, policy)
    if not isinstance(guarded_result, Mapping):
        guarded_result = {}
    return cls(
        kind=AgentEvidenceKind.TOOL,
        payload={
            "tool_identity": tool_identity,
            "tool_schema_name": tool_schema_name,
            "capture": capture.value,
            "result": guarded_result,
        },
        sensitive_fields=sensitive_fields
        if policy.include_sensitive_metadata
        else (),
        summary=summary,
    )

model_decision(*, model, decision, summary=None) classmethod

Create evidence metadata for a model decision.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def model_decision(
    cls,
    *,
    model: str,
    decision: JsonObject,
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create evidence metadata for a model decision."""
    _require_non_blank(model, "Agent model name")
    return cls(
        kind=AgentEvidenceKind.MODEL,
        payload={"model": model, "decision": decision},
        summary=summary,
    )

tool_decision(*, tool_identity, decision, summary=None) classmethod

Create evidence metadata for a tool-routing decision.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def tool_decision(
    cls,
    *,
    tool_identity: str,
    decision: JsonObject,
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create evidence metadata for a tool-routing decision."""
    _require_non_blank(tool_identity, "Agent tool identity")
    return cls(
        kind=AgentEvidenceKind.TOOL,
        payload={"tool_identity": tool_identity, "decision": decision},
        summary=summary,
    )

context_optimization(*, action, stage, signals=(), summary=None) classmethod

Create before/after evidence for a context optimization action.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def context_optimization(
    cls,
    *,
    action: ContextOptimizationAction,
    stage: ContextOptimizationEvidenceStage,
    signals: Sequence[ContextHealthSignal] = (),
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create before/after evidence for a context optimization action."""
    _require_non_blank(action.id, "Context optimization action id")
    return cls(
        kind=AgentEvidenceKind.CONTEXT_OPTIMIZATION,
        payload={
            "stage": stage.value,
            "action": action.evidence_payload(),
            "signals": tuple(signal.evidence_payload() for signal in signals),
        },
        summary=summary,
        digest=action.digest_ref,
        manifest_ref=action.manifest_ref,
        reference=action.result_evidence_ref,
    )

to_evidence(*, evidence_id, agent_state_id, created_at=None)

Assign repository identity while preserving append-only contents.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

def to_evidence(
    self,
    *,
    evidence_id: str,
    agent_state_id: str,
    created_at: datetime | None = None,
) -> AgentEvidence:
    """Assign repository identity while preserving append-only contents."""
    _require_non_blank(evidence_id, "Agent evidence id")
    _require_non_blank(agent_state_id, "Agent state id")
    return AgentEvidence(
        id=evidence_id,
        agent_state_id=agent_state_id,
        kind=self.kind,
        payload=self.payload,
        summary=self.summary,
        digest=self.digest,
        manifest_ref=self.manifest_ref,
        reference=self.reference,
        sensitive_fields=self.sensitive_fields,
        created_at=created_at,
    )

AgentEvidenceKind

Bases: StrEnum

Kinds of append-only evidence captured by agent execution.

AgentEventAttribution(agent_id, run_id, conversation_id, parent_run_id=None) dataclass

Attribution carried by every protocol-neutral agent event.

parent_run_id is None exactly when the run is a delegation-tree root (ADR-0013 §3 "parent link" / ADR-0009 parent linkage). The remaining ids are always present so adapters never have to synthesize linkage.

__post_init__()

Reject attribution that cannot identify an agent, run, or conversation.

Source code in core/spakky-agent/src/spakky/agent/event.py

def __post_init__(self) -> None:
    """Reject attribution that cannot identify an agent, run, or conversation."""
    if not self.agent_id.strip():
        raise AgentDefinitionError("Agent event agent id cannot be blank")
    if not self.run_id.strip():
        raise AgentDefinitionError("Agent event run id cannot be blank")
    if not self.conversation_id.strip():
        raise AgentDefinitionError("Agent event conversation id cannot be blank")
    if self.parent_run_id is not None and not self.parent_run_id.strip():
        raise AgentDefinitionError("Agent event parent run id cannot be blank")

AgentEventKind

Bases: StrEnum

Protocol-neutral event kinds emitted across an agent run.

The lifecycle kinds (RUN_STARTED … ARTIFACT) generalize ADR-0009's public AgentYield vocabulary and align 1:1 with the streaming events that AG-UI and A2A adapters must reproduce (ADR-0013 §3).

ArtifactEvent(attribution, artifact_id, content, name=None, metadata=dict()) dataclass

A produced artifact surfaced by an agent run.

MessageDeltaEvent(attribution, message_id, delta, metadata=dict()) dataclass

Incremental assistant message text produced by the model.

ReasoningDeltaEvent(attribution, reasoning_id, delta, metadata=dict()) dataclass

Incremental model reasoning (thinking) text.

RunFinishedEvent(attribution, error=None, metadata=dict()) dataclass

A run has finished executing.

error is None for a successful run and carries a terminal failure payload otherwise (A2A failed task / AG-UI RUN_ERROR).

RunPausedEvent(attribution, reason, prompt, state_id, approval_id=None, tool_call_id=None, allowed_decisions=(), metadata=dict()) dataclass

A run has paused without terminal success or failure.

The event is intentionally protocol-neutral: it carries core lifecycle reason plus the input prompt/decision envelope adapters need to project their own input-required or auth-required protocol states.

__post_init__()

Reject pause events that cannot be shown or resumed.

Source code in core/spakky-agent/src/spakky/agent/event.py

def __post_init__(self) -> None:
    """Reject pause events that cannot be shown or resumed."""
    if not self.prompt.strip():
        raise AgentDefinitionError("Agent run pause prompt cannot be blank")
    if not self.state_id.strip():
        raise AgentDefinitionError("Agent run pause state id cannot be blank")
    if self.approval_id is not None and not self.approval_id.strip():
        raise AgentDefinitionError("Agent run pause approval id cannot be blank")
    if self.tool_call_id is not None and not self.tool_call_id.strip():
        raise AgentDefinitionError("Agent run pause tool call id cannot be blank")

RunStartedEvent(attribution, metadata=dict()) dataclass

A run has started executing.

StateDeltaEvent(attribution, patch, metadata=dict()) dataclass

An incremental change to shared run state.

patch is a sequence of JSON-Patch operations (RFC 6902), the same shape AG-UI STATE_DELTA carries, so adapters relay it without re-deriving a diff.

StateSnapshotEvent(attribution, snapshot, metadata=dict()) dataclass

A full snapshot of shared run state.

StepFinishedEvent(attribution, step_name, metadata=dict()) dataclass

A named step inside a run has finished.

StepStartedEvent(attribution, step_name, metadata=dict()) dataclass

A named step inside a run has started (one model-loop iteration).

ToolCallArgsDeltaEvent(attribution, call_id, args_delta, metadata=dict()) dataclass

Incremental serialized arguments for an in-flight tool call.

ToolCallEndEvent(attribution, call_id, metadata=dict()) dataclass

A tool call has finished streaming its arguments.

ToolCallResultEvent(attribution, call_id, tool_name, message_id, result=None, metadata=dict()) dataclass

The result returned by executing a completed tool call.

message_id identifies the conversation message this tool result belongs to, so an adapter can attach the result to the right message. AG-UI requires it as messageId on TOOL_CALL_RESULT; without it the result frame cannot be reconstructed losslessly.

ToolCallStartEvent(attribution, call_id, tool_name, parent_message_id=None, metadata=dict()) dataclass

A model has begun emitting a tool call with a known name.

parent_message_id links the tool call to the assistant message that requested it (None when the model emits the call outside any message). AG-UI projects this as parentMessageId on TOOL_CALL_START.

AgentApprovalBoundaryKind

Bases: StrEnum

Action boundaries where orchestration may require human approval.

AgentApprovalDecisionOutcome(request_id, decision, status, transition, reason=None, modified_payload=dict(), comment=None) dataclass

Typed result of an approval decision signal.

AgentApprovalPlan(action, request=None, state=None, yield_item=None) dataclass

Plan for an action boundary before executing it.

requires_approval property

Return whether orchestration must wait for a HITL decision.

AgentApprovalPlanAction

Bases: StrEnum

Approval plan outcome before an action boundary is executed.

AgentApprovalRequest(id, agent_state_id, boundary, prompt, risk, action_ref, allowed_decisions=DEFAULT_APPROVAL_DECISIONS, metadata=dict()) dataclass

Approval request materialized at a risky action boundary.

__post_init__()

Reject approval requests that cannot be matched by a signal.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def __post_init__(self) -> None:
    """Reject approval requests that cannot be matched by a signal."""
    _require_non_blank(self.id, "Agent approval id")
    _require_non_blank(self.agent_state_id, "Agent approval state id")
    _require_non_blank(self.prompt, "Agent approval prompt")
    _require_non_blank(self.action_ref, "Agent approval action ref")
    if len(self.allowed_decisions) == 0:
        raise AgentDefinitionError("Agent approval decisions cannot be empty")

from_tool_descriptor(*, approval_id, agent_state_id, descriptor, prompt=None, action_ref=None, call_id=None, metadata=None) classmethod

Build an approval request from a risky tool descriptor.

Source code in core/spakky-agent/src/spakky/agent/approval.py

@classmethod
def from_tool_descriptor(
    cls,
    *,
    approval_id: str,
    agent_state_id: str,
    descriptor: AgentToolDescriptor,
    prompt: str | None = None,
    action_ref: str | None = None,
    call_id: str | None = None,
    metadata: JsonObject | None = None,
) -> "AgentApprovalRequest":
    """Build an approval request from a risky tool descriptor."""
    request_metadata = _tool_approval_metadata(descriptor, call_id, metadata or {})
    return cls(
        id=approval_id,
        agent_state_id=agent_state_id,
        boundary=AgentApprovalBoundaryKind.TOOL_INVOCATION,
        prompt=prompt or f"Approve tool invocation: {descriptor.name}",
        risk=descriptor.metadata.risk,
        action_ref=action_ref or descriptor.identity.key,
        metadata=request_metadata,
    )

to_state(*, agent_type)

Materialize the approval wait as interrupted lifecycle state.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def to_state(self, *, agent_type: str) -> AgentState:
    """Materialize the approval wait as interrupted lifecycle state."""
    _require_non_blank(agent_type, "Agent approval agent type")
    return AgentState(
        id=self.agent_state_id,
        agent_type=agent_type,
        status=AgentStatus.INTERRUPTED,
        transition=AgentStateTransition.WAITING_APPROVAL,
        reason=AgentStateReason.APPROVAL_REQUIRED,
        current_activity=self.prompt,
        metadata={"approval": self.to_metadata()},
    )

to_yield()

Expose this approval request to an inbound adapter stream.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def to_yield(self) -> AgentYield[Approval]:
    """Expose this approval request to an inbound adapter stream."""
    return AgentYield(
        kind=AgentYieldKind.APPROVAL,
        payload=Approval(
            id=self.id,
            prompt=self.prompt,
            allowed_decisions=self.allowed_decisions,
            metadata=self.to_metadata(),
        ),
    )

to_metadata()

Return JSON-compatible metadata for state, yield, and evidence.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def to_metadata(self) -> JsonObject:
    """Return JSON-compatible metadata for state, yield, and evidence."""
    return {
        "id": self.id,
        "agent_state_id": self.agent_state_id,
        "boundary": self.boundary.value,
        "action_ref": self.action_ref,
        "risk_axes": [axis.value for axis in self.risk.axes],
        "allowed_decisions": [
            decision.value for decision in self.allowed_decisions
        ],
        "metadata": self.metadata,
    }

AgentActionBoundaryCheckpoint(action_id, action_kind, stage, idempotency=Idempotency.UNKNOWN, metadata=dict()) dataclass

Serializable checkpoint recorded before or after one external action.

__post_init__()

Reject checkpoints that cannot be correlated after restart.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

def __post_init__(self) -> None:
    """Reject checkpoints that cannot be correlated after restart."""
    if not self.action_id.strip():
        raise AgentDefinitionError("Agent action boundary id cannot be blank")

before_model_call(action_id, *, idempotency=Idempotency.UNKNOWN, metadata=None) classmethod

Create the checkpoint recorded before a model call is attempted.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def before_model_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency = Idempotency.UNKNOWN,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded before a model call is attempted."""
    return cls._for_action(
        action_id,
        AgentActionKind.MODEL_CALL,
        AgentActionBoundaryStage.BEFORE,
        idempotency=idempotency,
        metadata=metadata,
    )

after_model_call(action_id, *, idempotency=Idempotency.UNKNOWN, metadata=None) classmethod

Create the checkpoint recorded after a model call completes.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def after_model_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency = Idempotency.UNKNOWN,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded after a model call completes."""
    return cls._for_action(
        action_id,
        AgentActionKind.MODEL_CALL,
        AgentActionBoundaryStage.AFTER,
        idempotency=idempotency,
        metadata=metadata,
    )

before_tool_call(action_id, *, idempotency, metadata=None) classmethod

Create the checkpoint recorded before a tool call is attempted.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def before_tool_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded before a tool call is attempted."""
    return cls._for_action(
        action_id,
        AgentActionKind.TOOL_CALL,
        AgentActionBoundaryStage.BEFORE,
        idempotency=idempotency,
        metadata=metadata,
    )

after_tool_call(action_id, *, idempotency, metadata=None) classmethod

Create the checkpoint recorded after a tool call completes.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def after_tool_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded after a tool call completes."""
    return cls._for_action(
        action_id,
        AgentActionKind.TOOL_CALL,
        AgentActionBoundaryStage.AFTER,
        idempotency=idempotency,
        metadata=metadata,
    )

before_approval_wait(action_id, *, metadata=None) classmethod

Create the checkpoint recorded before waiting for approval.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def before_approval_wait(
    cls,
    action_id: str,
    *,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded before waiting for approval."""
    return cls._for_action(
        action_id,
        AgentActionKind.APPROVAL_WAIT,
        AgentActionBoundaryStage.BEFORE,
        idempotency=Idempotency.NON_IDEMPOTENT,
        metadata=metadata,
    )

after_approval_wait(action_id, *, metadata=None) classmethod

Create the checkpoint recorded after an approval wait resolves.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def after_approval_wait(
    cls,
    action_id: str,
    *,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded after an approval wait resolves."""
    return cls._for_action(
        action_id,
        AgentActionKind.APPROVAL_WAIT,
        AgentActionBoundaryStage.AFTER,
        idempotency=Idempotency.NON_IDEMPOTENT,
        metadata=metadata,
    )

to_evidence_candidate(*, summary=None)

Represent this checkpoint as append-only evidence.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

def to_evidence_candidate(
    self,
    *,
    summary: str | None = None,
) -> AgentEvidenceCandidate:
    """Represent this checkpoint as append-only evidence."""
    return AgentEvidenceCandidate(
        kind=AgentEvidenceKind.ACTION_BOUNDARY,
        payload={
            "action_id": self.action_id,
            "action_kind": self.action_kind.value,
            "stage": self.stage.value,
            "idempotency": self.idempotency.value,
            "metadata": self.metadata,
        },
        summary=summary,
    )

AgentActionBoundaryStage

Bases: StrEnum

Checkpoint side recorded around an action boundary.

AgentActionKind

Bases: StrEnum

Recoverable external action classes in an agent execution.

AgentResumeAction

Bases: StrEnum

Orchestration action selected from persisted checkpoint evidence.

AgentResumeBoundary(action_id, action_kind, stage, idempotency, evidence_id) dataclass

Last action boundary reconstructed from append-only evidence.

AgentResumePlan(state, action, boundary=None, signals=()) dataclass

Resume decision derived from persisted state and checkpoint evidence.

requires_human_input property

Return whether automatic replay must stop at HITL recovery.

can_resume_automatically property

Return whether orchestration may continue without approval.

AgentDelegateTarget(agent_type, agent_name=None, metadata=dict()) dataclass

First-class delegate target represented by another @Agent component.

__post_init__()

Reject delegate targets that cannot identify an agent component.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegate targets that cannot identify an agent component."""
    if not self.agent_type.strip():
        raise AgentDefinitionError("Delegate target agent type cannot be blank")
    if self.agent_name is not None and not self.agent_name.strip():
        raise AgentDefinitionError("Delegate target agent name cannot be blank")

DelegationToolResult(summary, output=None, events=(), metadata=dict()) dataclass

Model-facing result plus child neutral events from a teammate call.

__post_init__()

Reject delegated tool results without a model-facing summary.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegated tool results without a model-facing summary."""
    if not self.summary.strip():
        raise AgentDefinitionError("Delegation tool result summary cannot be blank")

from_result(result, *, events=()) classmethod

Build a model-facing tool result from a delegation result object.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

@classmethod
def from_result(
    cls,
    result: DelegationResult,
    *,
    events: tuple[AgentEvent, ...] = (),
) -> "DelegationToolResult":
    """Build a model-facing tool result from a delegation result object."""
    return cls(
        summary=result.summary,
        output=result.output,
        events=events,
        metadata={
            "delegation_id": result.id,
            "packet_id": result.packet_id,
            "target_agent_type": result.target.agent_type,
            **result.metadata,
        },
    )

DelegationBudget(max_steps=None, max_tokens=None, timeout_seconds=None, deadline_at=None, metadata=dict()) dataclass

Budget metadata attached to a delegation packet.

__post_init__()

Reject delegation budgets that cannot be enforced consistently.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegation budgets that cannot be enforced consistently."""
    if self.max_steps is not None and self.max_steps <= 0:
        raise AgentDefinitionError("Delegation max steps must be positive")
    if self.max_tokens is not None and self.max_tokens <= 0:
        raise AgentDefinitionError("Delegation max tokens must be positive")
    if self.timeout_seconds is not None and self.timeout_seconds <= 0:
        raise AgentDefinitionError("Delegation timeout must be positive")

DelegationContextSlice(summary=None, evidence_refs=(), manifest_ref=None, metadata=dict()) dataclass

Minimal parent context projected for a child agent.

DelegationExpectedOutput(description=None, schema=dict(), metadata=dict()) dataclass

Expected child output description and optional JSON schema.

DelegationPacket(id, parent_agent_state_id, target, task, context=DelegationContextSlice(), constraints=(), expected_output=DelegationExpectedOutput(), budget=DelegationBudget(), allowed_capabilities=(), return_policy=DelegationReturnPolicy.SUMMARY_AND_EVIDENCE, metadata=dict()) dataclass

Task packet passed from a parent agent to a delegate agent.

__post_init__()

Reject delegation packets without parent linkage or task identity.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegation packets without parent linkage or task identity."""
    if not self.id.strip():
        raise AgentDefinitionError("Delegation id cannot be blank")
    if not self.parent_agent_state_id.strip():
        raise AgentDefinitionError("Delegation parent state id cannot be blank")

DelegationResult(id, packet_id, target, summary, output=None, evidence_refs=(), metadata=dict(), created_at=None) dataclass

Child agent result projected back to the parent execution.

__post_init__()

Reject delegated results that cannot be linked to a packet.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegated results that cannot be linked to a packet."""
    if not self.id.strip():
        raise AgentDefinitionError("Delegation result id cannot be blank")
    if not self.packet_id.strip():
        raise AgentDefinitionError("Delegation result packet id cannot be blank")
    if not self.summary.strip():
        raise AgentDefinitionError("Delegation result summary cannot be blank")

to_parent_evidence(*, evidence_id, parent_agent_state_id)

Represent a delegated result as append-only parent evidence.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def to_parent_evidence(
    self,
    *,
    evidence_id: str,
    parent_agent_state_id: str,
) -> AgentEvidence:
    """Represent a delegated result as append-only parent evidence."""
    payload: dict[str, JsonValue] = {
        "delegation_id": self.id,
        "packet_id": self.packet_id,
        "target_agent_type": self.target.agent_type,
        "evidence_refs": self.evidence_refs,
        "metadata": self.metadata,
    }
    if self.target.agent_name is not None:
        payload["target_agent_name"] = self.target.agent_name
    if self.output is not None:
        payload["output"] = self.output
    return AgentEvidence(
        id=evidence_id,
        agent_state_id=parent_agent_state_id,
        kind=AgentEvidenceKind.DELEGATION,
        payload=payload,
        summary=self.summary,
        created_at=self.created_at,
    )

to_parent_yield(*, evidence_id, parent_agent_state_id)

Expose the delegated result on the parent's AgentYield stream.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def to_parent_yield(
    self,
    *,
    evidence_id: str,
    parent_agent_state_id: str,
) -> AgentYield[Evidence]:
    """Expose the delegated result on the parent's AgentYield stream."""
    evidence = self.to_parent_evidence(
        evidence_id=evidence_id,
        parent_agent_state_id=parent_agent_state_id,
    )
    return AgentYield(
        kind=AgentYieldKind.EVIDENCE,
        payload=Evidence(
            evidence=evidence,
            metadata={"delegation_id": self.id, "packet_id": self.packet_id},
        ),
    )

DelegationReturnPolicy

Bases: StrEnum

How a child agent result should be projected back to the parent.

IAgentDelegate

Bases: ABC

Execution hook that runs a delegation packet against a delegate target.

delegate(packet) abstractmethod

Execute delegation without prescribing spawn topology or transport.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

@abstractmethod
def delegate(
    self,
    packet: DelegationPacket,
) -> AsyncGenerator[AgentYield[DelegationResult], None]:
    """Execute delegation without prescribing spawn topology or transport."""

delegate_tool_result(packet) async

Execute delegation and collect the terminal result for a tool call.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

async def delegate_tool_result(
    self,
    packet: DelegationPacket,
) -> DelegationToolResult:
    """Execute delegation and collect the terminal result for a tool call."""
    terminal: DelegationResult | None = None
    async for item in self.delegate(packet):
        if isinstance(item.payload, DelegationResult):
            terminal = item.payload
    if terminal is None:
        raise AgentToolDispatchError("Agent delegate produced no delegation result")
    return DelegationToolResult.from_result(terminal)

ICompactionStrategy

Bases: ABC

Provider-neutral context-compaction transform applied before a request.

A strategy maps a resolved history to a shorter one, reading the running ModelUsage and backend ModelCapability so it can scale its effect to how close the run is to the context limit. Implementations are pure transforms — they never mutate the input — so the runner can apply a declared chain by threading each strategy's output into the next.

compact(history, usage, capability) abstractmethod async

Return a compacted view of history for the next model request.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

@abstractmethod
async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Return a compacted view of ``history`` for the next model request."""
    ...

KeepRecentMessagesCompactionStrategy(max_messages) dataclass

Bases: ICompactionStrategy

Sliding-window strategy that keeps only the most recent messages.

The cheapest compaction: drop the oldest messages and keep the last max_messages. It carries no model dependency, so it is the safe default tail of a chain that bounds history length regardless of content.

__post_init__()

Reject a window that would keep no messages.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

def __post_init__(self) -> None:
    """Reject a window that would keep no messages."""
    if self.max_messages <= 0:
        raise AgentDefinitionError(
            "Keep-recent compaction window must keep at least one message"
        )

compact(history, usage, capability) async

Keep only the last max_messages messages of the history.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Keep only the last ``max_messages`` messages of the history."""
    return history[-self.max_messages :]

ProviderManagedCompactionStrategy() dataclass

Bases: ICompactionStrategy

No-op strategy for backends that manage their own context window.

Some providers compact server-side, so the framework must not also trim the transcript. Declaring this strategy makes that hand-off explicit in the spec rather than leaving compaction silently absent.

compact(history, usage, capability) async

Return the history unchanged — the provider owns compaction.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Return the history unchanged — the provider owns compaction."""
    return history

SummarizeOldTurnsCompactionStrategy(model, keep_recent, summary_instruction=DEFAULT_SUMMARY_INSTRUCTION) dataclass

Bases: ICompactionStrategy

Replace older turns with a model-generated summary, keeping recent ones.

The richest compaction: a secondary model call condenses the turns older than keep_recent into one EVIDENCE summary message that precedes the kept tail. This preserves earlier context as a briefing instead of dropping it, at the cost of one extra model round-trip. When the history is already within keep_recent there is nothing older to summarize, so the call is skipped.

__post_init__()

Reject a window that would leave no recent turns to anchor the summary.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

def __post_init__(self) -> None:
    """Reject a window that would leave no recent turns to anchor the summary."""
    if self.keep_recent <= 0:
        raise AgentDefinitionError(
            "Summarize compaction must keep at least one recent message"
        )
    if not self.summary_instruction.strip():
        raise AgentDefinitionError(
            "Summarize compaction instruction cannot be blank"
        )

compact(history, usage, capability) async

Summarize turns older than keep_recent ahead of the recent tail.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Summarize turns older than ``keep_recent`` ahead of the recent tail."""
    if len(history) <= self.keep_recent:
        return history
    older = history[: len(history) - self.keep_recent]
    recent = history[len(history) - self.keep_recent :]
    summary = await self.model.complete(self._summary_request(older))
    return (self._summary_message(summary.content), *recent)

TrimToolResultsCompactionStrategy(max_characters) dataclass

Bases: ICompactionStrategy

Truncate verbose tool-result messages while preserving the dialogue.

Tool results (search dumps, file contents) dominate token cost yet rarely need to be replayed verbatim. This strategy truncates only TOOL role message content past max_characters and leaves the user/assistant turns untouched, so the model still sees that a tool ran and a clipped head of its output.

__post_init__()

Reject a non-positive truncation budget.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

def __post_init__(self) -> None:
    """Reject a non-positive truncation budget."""
    if self.max_characters <= 0:
        raise AgentDefinitionError(
            "Tool-result trim budget must be a positive character count"
        )

compact(history, usage, capability) async

Truncate over-budget tool-result content, leaving other roles intact.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Truncate over-budget tool-result content, leaving other roles intact."""
    return tuple(self._trim(message) for message in history)

Agent(spec=AgentExecutionSpec(), *, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

UseCase-equivalent Pod stereotype for agentic workflow components.

validate_bootstrap()

Re-run definition validation during application bootstrap.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def validate_bootstrap(self) -> None:
    """Re-run definition validation during application bootstrap."""
    self._validate_execute_contract(self._ensure_agent_class(self.target))

required_persistence_repository_types()

Return repository ports required by this Agent's durable path.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def required_persistence_repository_types(self) -> tuple[type[object], ...]:
    """Return repository ports required by this Agent's durable path."""
    from spakky.agent.interfaces.repository import (
        IAgentEvidenceRepository,
        IAgentSignalRepository,
        IAgentStateRepository,
    )

    if (
        self.spec.recovery is RecoveryStrategy.ACTION_BOUNDARY
        or len(self.spec.accepted_signals) > 0
    ):
        return (
            IAgentStateRepository,
            IAgentSignalRepository,
            IAgentEvidenceRepository,
        )
    return ()

AgentCompactionPolicy(strategies, trigger_token_threshold) dataclass

Declared compaction chain plus the token threshold that triggers it.

The strategies form an ordered chain of pluggable ICompactionStrategy ports applied in sequence once the running token estimate crosses trigger_token_threshold. The runner threads each strategy's output into the next, so chain order is the compaction order (ADR-0013 §7).

__post_init__()

Reject compaction policies that cannot be enforced consistently.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject compaction policies that cannot be enforced consistently."""
    if not self.strategies:
        raise AgentDefinitionError(
            "Agent compaction policy requires at least one strategy"
        )
    if self.trigger_token_threshold <= 0:
        raise AgentDefinitionError(
            "Agent compaction trigger token threshold must be positive"
        )

AgentExecutionLimits(timeout_seconds=None) dataclass

Bounded execution limits declared outside infrastructure capabilities.

__post_init__()

Reject limits that would fail later at bootstrap.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject limits that would fail later at bootstrap."""
    if self.timeout_seconds is not None and self.timeout_seconds <= 0:
        raise AgentDefinitionError("Agent timeout limit must be positive")

AgentExecutionSpec(name=None, objective=None, instructions=None, output_type=None, accepted_signals=(), recovery=RecoveryStrategy.NONE, streaming_exposure_mode=StreamingExposureMode.BALANCED, timeout_seconds=None, limits=AgentExecutionLimits(), teammates=(), compaction=None, delegation_allowed=False, metadata=dict()) dataclass

Declarative execution semantics that cannot be inferred from DI alone.

__post_init__()

Reject execution specs that would fail later at bootstrap.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject execution specs that would fail later at bootstrap."""
    if self.name is not None and not self.name.strip():
        raise AgentDefinitionError("Agent name cannot be blank")
    if self.objective is not None and not self.objective.strip():
        raise AgentDefinitionError("Agent objective cannot be blank")
    if self.instructions is not None and not self.instructions.strip():
        raise AgentDefinitionError("Agent instructions cannot be blank")
    if self.output_type is not None and not isclass(self.output_type):
        raise AgentDefinitionError("Agent output type must be a class")
    if self.timeout_seconds is not None and self.timeout_seconds <= 0:
        raise AgentDefinitionError("Agent timeout must be positive")
    if (
        self.timeout_seconds is not None
        and self.limits.timeout_seconds is not None
        and self.timeout_seconds != self.limits.timeout_seconds
    ):
        raise AgentDefinitionError("Agent timeout declarations must match")
    teammate_names = [teammate.name for teammate in self.teammates]
    if len(set(teammate_names)) != len(teammate_names):
        raise AgentDefinitionError("Agent teammate names must be unique")

AgentTeammate(name, pod=None, card_url=None) dataclass

Declared collaborator an agent may delegate to during execution.

A teammate is resolved either in-process by a local @Agent Pod type or remotely by an AgentCard endpoint URL. Exactly one binding is declared; the runtime exposes it as a synthetic teammate delegation tool and dispatches it through the local runner or an injected IAgentDelegate port.

__post_init__()

Reject teammates that cannot resolve to a single delegate target.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject teammates that cannot resolve to a single delegate target."""
    if not self.name.strip():
        raise AgentDefinitionError("Agent teammate name cannot be blank")
    if (self.pod is None) == (self.card_url is None):
        raise AgentDefinitionError(
            "Agent teammate must declare exactly one of a local pod "
            "or a remote AgentCard url"
        )
    if self.pod is not None and not isclass(self.pod):
        raise AgentDefinitionError("Agent teammate pod must be a class")
    if self.card_url is not None:
        parts = urlsplit(self.card_url)
        if parts.scheme not in REMOTE_AGENT_CARD_SCHEMES or not parts.netloc:
            raise AgentDefinitionError(
                "Agent teammate AgentCard url must be an http(s) endpoint"
            )

RecoveryStrategy

Bases: StrEnum

Recovery strategy requested by an agent execution contract.

StreamingExposureMode

Bases: StrEnum

Streaming output guard profile exposed by agent execution.

AgentSignalHookCatalog(descriptors=()) dataclass

Deterministic catalog of signal hooks discovered from an Agent class.

__post_init__()

Reject duplicate hook identities so dispatch order stays stable.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def __post_init__(self) -> None:
    """Reject duplicate hook identities so dispatch order stays stable."""
    identity_keys: set[str] = set()
    for descriptor in self.descriptors:
        if descriptor.identity.key in identity_keys:
            raise AgentDefinitionError("Agent signal hook identity must be unique")
        identity_keys.add(descriptor.identity.key)

hooks_for(kind)

Return every hook that handles the requested signal kind, in order.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def hooks_for(
    self,
    kind: AgentSignalKind,
) -> tuple[AgentSignalHookDescriptor, ...]:
    """Return every hook that handles the requested signal kind, in order."""
    return tuple(
        descriptor for descriptor in self.descriptors if descriptor.kind is kind
    )

AgentSignalHookDescriptor(identity, owner, callable, kind) dataclass

Discovered signal hook bound to an owner class and callable.

AgentSignalHookIdentity(owner_module, owner_qualname, member_name) dataclass

Hook identity independent from the developer-chosen method name.

key property

Return a stable key for deterministic ordering and logs.

RunAgentInput(state_id, instruction, conversation_id=None, parent_run_id=None, resume=False, message_history=(), model_selection=None, metadata=dict()) dataclass

Inbound contract for one framework-owned agent run.

effective_conversation_id property

Return the multi-turn thread id, defaulting to the run id.

ADR-0013 §3 attribution requires a conversation id on every event. A single-turn caller may omit it, in which case the run id identifies the (degenerate) one-turn conversation.

__post_init__()

Reject inbound input that cannot correlate a run or seed a request.

Source code in core/spakky-agent/src/spakky/agent/inbound.py

def __post_init__(self) -> None:
    """Reject inbound input that cannot correlate a run or seed a request."""
    if not self.state_id.strip():
        raise AgentDefinitionError("Run agent input state id cannot be blank")
    if not self.instruction.strip():
        raise AgentDefinitionError("Run agent input instruction cannot be blank")
    if self.conversation_id is not None and not self.conversation_id.strip():
        raise AgentDefinitionError(
            "Run agent input conversation id cannot be blank"
        )
    if self.parent_run_id is not None and not self.parent_run_id.strip():
        raise AgentDefinitionError("Run agent input parent run id cannot be blank")

IAgentModelResolver

Bases: ABC

Resolve the model adapter used for one Agent run.

resolve_model(agent_instance, run_input=None) abstractmethod

Return a run-specific model, or None to use injected fallback.

Source code in core/spakky-agent/src/spakky/agent/model_resolver.py

@abstractmethod
def resolve_model(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> IAgentModel | None:
    """Return a run-specific model, or ``None`` to use injected fallback."""
    ...

AgentRunner(agent, target, model, states=None, signals=None, evidence=None, task_store=None) dataclass

Framework-owned standard agent loop bound to one agent instance.

Durable repositories are None for a stateless agent (one that declares no accepted_signals and no action-boundary recovery). In that mode the runner skips state/evidence/signal handling and runs model -> tool -> final.

for_agent_instance(instance) classmethod

Resolve runner ports from an agent instance's injected attributes.

ADR-0009 stores constructor-injected ports as instance attributes, so the runner reads vars(instance) (the typed instance __dict__, not the banned getattr) and resolves each port by runtime type. Attribute names are developer-chosen, so resolution is type-driven, not name-driven.

Source code in core/spakky-agent/src/spakky/agent/runner.py

@classmethod
def for_agent_instance(cls, instance: object) -> "AgentRunner":
    """Resolve runner ports from an agent instance's injected attributes.

    ADR-0009 stores constructor-injected ports as instance attributes, so the
    runner reads ``vars(instance)`` (the typed instance ``__dict__``, not the
    banned ``getattr``) and resolves each port by runtime type. Attribute
    names are developer-chosen, so resolution is type-driven, not name-driven.
    """
    agent = Agent.get(type(instance))
    attributes = tuple(vars(instance).values())
    model = cls._resolve_required(attributes, IAgentModel)
    if model is None:
        raise AgentModelConfigurationError(
            "Agent run requires an IAgentModel port but none was injected"
        )
    states = cls._resolve_optional(attributes, IAgentStateRepository)
    signals = cls._resolve_optional(attributes, IAgentSignalRepository)
    evidence = cls._resolve_optional(attributes, IAgentEvidenceRepository)
    task_store = cls._resolve_optional(attributes, ITaskStore)
    runner = cls(
        agent=agent,
        target=instance,
        model=model,
        states=states,
        signals=signals,
        evidence=evidence,
        task_store=task_store,
    )
    runner._require_durable_ports()
    return runner

with_model(model)

Return a runner using a run-specific model adapter.

Source code in core/spakky-agent/src/spakky/agent/runner.py

def with_model(self, model: IAgentModel) -> "AgentRunner":
    """Return a runner using a run-specific model adapter."""
    return replace(self, model=model)

run(run_input) async

Run one model-mediated agent loop, yielding the public stream.

Source code in core/spakky-agent/src/spakky/agent/runner.py

async def run(
    self,
    run_input: RunAgentInput,
) -> AsyncGenerator[AgentYield[object], None]:
    """Run one model-mediated agent loop, yielding the public stream."""
    if self.states is None:
        async for item in self._run_stateless(run_input):
            yield item
        return
    async for item in self._run_durable(run_input, self.states):
        yield item

run_events(run_input) async

Run one model-mediated loop, emitting the neutral event taxonomy.

This is the lossless native stream AG-UI/A2A adapters consume (ADR-0013 §3). It shares the run's orchestration with run() — model request, tool dispatch, durable approval gating, evidence persistence — but emits distinct AgentEvent members instead of coarse AgentYield items, so an adapter projects each event one-to-one rather than re-expanding framing.

Source code in core/spakky-agent/src/spakky/agent/runner.py

async def run_events(
    self,
    run_input: RunAgentInput,
) -> AsyncGenerator[AgentEvent, None]:
    """Run one model-mediated loop, emitting the neutral event taxonomy.

    This is the lossless native stream AG-UI/A2A adapters consume (ADR-0013
    §3). It shares the run's orchestration with ``run()`` — model request,
    tool dispatch, durable approval gating, evidence persistence — but emits
    distinct ``AgentEvent`` members instead of coarse ``AgentYield`` items, so
    an adapter projects each event one-to-one rather than re-expanding framing.
    """
    attribution = self._event_attribution(run_input)
    state = (
        self._ensure_active_state(run_input, self.states)
        if self.states is not None
        else None
    )
    cursor = _MessageCursor(state_id=run_input.state_id)
    history = await self._resolved_history(run_input)
    assistant_text: list[str] = []
    yield RunStartedEvent(attribution)
    if state is not None:
        cancel = await self._consume_cancel(state)
        if cancel is not None:
            yield RunFinishedEvent(attribution, error=_cancel_error(cancel))
            return
    yield StepStartedEvent(attribution, step_name="model-call")
    async for event in self.model.stream(self._model_request(run_input, history)):
        if state is not None:
            cancel = await self._consume_cancel(state)
            if cancel is not None:
                yield StepFinishedEvent(attribution, step_name="model-call")
                yield RunFinishedEvent(attribution, error=_cancel_error(cancel))
                return
        async for item in self._consume_stream_event_as_events(
            event, state, attribution, cursor, run_input
        ):
            yield item
        self._accumulate_assistant_text(event, assistant_text)
        if event.kind is ModelStreamEventKind.ERROR and event.error is not None:
            yield StepFinishedEvent(attribution, step_name="model-call")
            yield RunFinishedEvent(
                attribution,
                error={"code": event.error.code, "message": event.error.message},
            )
            return
        if (
            state is not None
            and event.kind is ModelStreamEventKind.TOOL_CALL_CANDIDATE
            and event.tool_call is not None
            and self._states_required().get(state.id).status
            is not AgentStatus.ACTIVE
        ):
            current = self._states_required().get(state.id)
            yield StepFinishedEvent(attribution, step_name="model-call")
            if current.status is AgentStatus.INTERRUPTED:
                yield _run_paused_event(current, attribution)
                return
            yield RunFinishedEvent(attribution, error=_state_error(current))
            return
    yield StepFinishedEvent(attribution, step_name="model-call")
    # A tool that paused for approval already saved a WAIT_FOR_APPROVAL state
    # (INTERRUPTED, not ACTIVE); completing it would clobber the durable pause
    # the adapter surfaces as its deferred-tool request, so only an unpaused
    # (still ACTIVE) run transitions to COMPLETED — mirroring _run_durable.
    if (
        state is not None
        and self._states_required().get(state.id).status is AgentStatus.ACTIVE
    ):
        self._complete_state(state.id)
    elif state is not None:
        current = self._states_required().get(state.id)
        if current.status is AgentStatus.INTERRUPTED:
            yield _run_paused_event(current, attribution)
            return
        yield RunFinishedEvent(attribution, error=_state_error(current))
        return
    self._persist_turns(run_input, assistant_text)
    yield RunFinishedEvent(attribution)

AgentRunResult(state_id, status, tool_calls, evidence_count) dataclass

Neutral terminal summary returned when a spec declares no output type.

AgentRunnerFactory(model_resolver=None)

Bases: IAgentRunnerFactory

Default runner factory using the native framework-owned runner.

Source code in core/spakky-agent/src/spakky/agent/runner_factory.py

def __init__(self, model_resolver: IAgentModelResolver | None = None) -> None:
    self._model_resolver = model_resolver

open_runner(agent_instance, run_input=None) async

Yield the native runner for one request.

Source code in core/spakky-agent/src/spakky/agent/runner_factory.py

@override
@asynccontextmanager
async def open_runner(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> AsyncGenerator[AgentRunner, None]:
    """Yield the native runner for one request."""
    runner = AgentRunner.for_agent_instance(agent_instance)
    if self._model_resolver is not None:
        model = self._model_resolver.resolve_model(agent_instance, run_input)
        if model is not None:
            runner = runner.with_model(model)
    yield runner

IAgentRunnerFactory

Bases: ABC

Port for opening a request-scoped runner for one agent instance.

open_runner(agent_instance, run_input=None) abstractmethod

Yield a runner bound to agent_instance for one adapter request.

Source code in core/spakky-agent/src/spakky/agent/runner_factory.py

@abstractmethod
def open_runner(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> AbstractAsyncContextManager[AgentRunner]:
    """Yield a runner bound to ``agent_instance`` for one adapter request."""
    ...

IAgentEvidenceRepository

Bases: ABC

Append-only evidence repository exposed to agent-facing code.

append(evidence) abstractmethod

Append immutable evidence and return the stored artifact.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def append(self, evidence: AgentEvidence) -> AgentEvidence:
    """Append immutable evidence and return the stored artifact."""
    ...

get(evidence_id) abstractmethod

Return one evidence artifact by id.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get(self, evidence_id: str) -> AgentEvidence:
    """Return one evidence artifact by id."""
    ...

list_by_state(state_id) abstractmethod

Return evidence artifacts captured for an agent state.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_state(self, state_id: str) -> Sequence[AgentEvidence]:
    """Return evidence artifacts captured for an agent state."""
    ...

list_by_manifest_ref(manifest_ref) abstractmethod

Return evidence artifacts associated with a context manifest.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_manifest_ref(self, manifest_ref: str) -> Sequence[AgentEvidence]:
    """Return evidence artifacts associated with a context manifest."""
    ...

IAgentSignalRepository

Bases: ABC

Durable inbound queue repository for agent signals.

append(signal) abstractmethod

Append a signal to the inbound queue and return the stored entry.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def append(self, signal: AgentSignal) -> AgentSignal:
    """Append a signal to the inbound queue and return the stored entry."""
    ...

list_pending(state_id) abstractmethod

Return unconsumed signals for an agent state in append/queue order.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_pending(self, state_id: str) -> Sequence[AgentSignal]:
    """Return unconsumed signals for an agent state in append/queue order."""
    ...

mark_consumed(signal_id) abstractmethod

Mark one queued signal as consumed after a non-blocking poll.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def mark_consumed(self, signal_id: str) -> AgentSignal:
    """Mark one queued signal as consumed after a non-blocking poll."""
    ...

IAgentStateRepository

Bases: ABC

Materialized state repository for long-running agent executions.

get(state_id) abstractmethod

Return a persisted agent state by id.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get(self, state_id: str) -> AgentState:
    """Return a persisted agent state by id."""
    ...

get_or_none(state_id) abstractmethod

Return a persisted agent state by id, or None when absent.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get_or_none(self, state_id: str) -> AgentState | None:
    """Return a persisted agent state by id, or None when absent."""
    ...

save(state) abstractmethod

Persist the materialized state and return the saved snapshot.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def save(self, state: AgentState) -> AgentState:
    """Persist the materialized state and return the saved snapshot."""
    ...

list_by_status(status) abstractmethod

Return states matching an externally observable lifecycle status.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_status(self, status: AgentStatus) -> Sequence[AgentState]:
    """Return states matching an externally observable lifecycle status."""
    ...

list_resume_candidates() abstractmethod

Return active or interrupted states that may resume after restart.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_resume_candidates(self) -> Sequence[AgentState]:
    """Return active or interrupted states that may resume after restart."""
    ...

ConversationTurn(role, content, metadata=dict()) dataclass

One persisted exchange in a multi-turn conversation transcript.

A turn is the protocol-neutral unit a TaskStore persists: who spoke (role) and what was said (content). It is intentionally narrower than a model-request ModelMessage — the transcript records the user/assistant dialogue that seeds future turns, not the system or evidence framing the runner assembles fresh on each request. as_model_message projects a turn back into the model-request vocabulary when the runner replays history.

__post_init__()

Reject turns that cannot seed a future model request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

def __post_init__(self) -> None:
    """Reject turns that cannot seed a future model request."""
    if self.role not in (ModelMessageRole.USER, ModelMessageRole.ASSISTANT):
        raise AgentDefinitionError(
            "Conversation turn role must be user or assistant"
        )
    if not self.content.strip():
        raise AgentDefinitionError("Conversation turn content cannot be blank")

as_model_message()

Project this turn into a model-request message for history replay.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

def as_model_message(self) -> ModelMessage:
    """Project this turn into a model-request message for history replay."""
    return ModelMessage(
        role=self.role, content=self.content, metadata=self.metadata
    )

ITaskStore

Bases: ABC

Durable conversation-history store keyed by conversation_id.

Persists the running transcript of a server-side session so a later run with the same conversation_id continues the conversation (ADR-0013 §6). A2A can map its contextId to this key for transcript replay, while A2A task snapshots are stored by spakky-a2a's repository bridge.

load_history(conversation_id) abstractmethod

Return the persisted transcript for a conversation in turn order.

Returns an empty sequence for a conversation that has no persisted turns yet — the first turn of a brand-new conversation.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

@abstractmethod
def load_history(self, conversation_id: str) -> Sequence[ConversationTurn]:
    """Return the persisted transcript for a conversation in turn order.

    Returns an empty sequence for a conversation that has no persisted turns
    yet — the first turn of a brand-new conversation.
    """
    ...

append_turns(conversation_id, turns) abstractmethod

Append new turns to a conversation's transcript in order.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

@abstractmethod
def append_turns(
    self,
    conversation_id: str,
    turns: Sequence[ConversationTurn],
) -> None:
    """Append new turns to a conversation's transcript in order."""
    ...

IAgentModel

Bases: ABC

Outbound model adapter port owned by spakky-agent core.

capability abstractmethod property

Return the backend capability descriptor queryable before a run.

capability_for(selection=None)

Return capability for a run-specific model selection.

Existing fixed-model adapters can ignore the selector and inherit the default. Routing adapters can override this to expose per-model context windows, reasoning support, or token-counting support before a request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def capability_for(
    self,
    selection: ModelSelection | None = None,
) -> ModelCapability:
    """Return capability for a run-specific model selection.

    Existing fixed-model adapters can ignore the selector and inherit the
    default. Routing adapters can override this to expose per-model context
    windows, reasoning support, or token-counting support before a request.
    """
    _ = selection
    return self.capability

complete(request) abstractmethod async

Return a complete model response for the request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

@abstractmethod
async def complete(self, request: ModelRequest) -> ModelResponse:
    """Return a complete model response for the request."""
    ...

stream(request) abstractmethod

Return provider-neutral stream events for the request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

@abstractmethod
def stream(self, request: ModelRequest) -> AsyncIterator[ModelStreamEvent]:
    """Return provider-neutral stream events for the request."""
    ...

JsonSchemaConstraint(schema, strict=True) dataclass

JSON schema constraint shared by structured output and tool calling.

ModelCapability(supports_reasoning=False, context_window_tokens=None, supports_token_counting=False) dataclass

Provider-neutral declaration of a model backend's queryable abilities.

The agent runner consults this descriptor before a run to adjust behaviour without invoking the backend. supports_reasoning gates whether the runner expects REASONING_DELTA events; when False the adapter omits them rather than failing (graceful degrade). context_window_tokens is None when the backend does not declare a fixed limit. supports_token_counting declares whether the backend can report token accounting for a request before sending it.

ModelError(code, message, retryable=False, metadata=dict()) dataclass

Provider-neutral model failure payload.

ModelMessage(role, content, metadata=dict()) dataclass

Provider-neutral model message.

ModelMessageRole

Bases: StrEnum

Roles accepted by provider-neutral model messages.

ModelRequest(messages, context=tuple(), context_manifest=None, context_digest=None, structured_output=None, tool_calling=None, sampling=SamplingOptions(), streaming=StreamingOptions(), model_selection=None, metadata=dict()) dataclass

Provider-neutral request passed to an agent model adapter.

assemble_messages(policy=None)

Assemble prompt messages from typed context packs without concatenation.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def assemble_messages(
    self,
    policy: ContextExposurePolicy | None = None,
) -> tuple[ModelMessage, ...]:
    """Assemble prompt messages from typed context packs without concatenation."""
    exposure_policy = policy or ContextExposurePolicy()
    context_messages = tuple(
        ModelMessage(
            role=ModelMessageRole.EVIDENCE,
            content=pack.guarded_content(exposure_policy),
            metadata=pack.message_metadata(exposure_policy),
        )
        for pack in self.context
    )
    return (*self.messages, *context_messages)

ModelResponse(content, structured_output=None, tool_calls=tuple(), usage=ModelUsage(), metadata=dict()) dataclass

Provider-neutral non-streaming model response.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive output payloads deterministically guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelResponse":
    """Return a copy with sensitive output payloads deterministically guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    content = self.content
    structured_output = self.structured_output
    content_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path in ((), ("content",))
    )
    if content_descriptors:
        guarded_content = guard_json_value(
            {"content": content},
            tuple(
                SensitiveFieldDescriptor(("content",), descriptor.field)
                for descriptor in content_descriptors
            ),
            exposure_policy,
        )
        content_value = cast(Mapping[str, JsonValue], guarded_content).get(
            "content"
        )
        if isinstance(content_value, str):
            content = content_value
        else:
            content = "[REDACTED]"
    structured_output_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path not in ((), ("content",))
    )
    if structured_output_descriptors:
        structured_output = guard_json_value(
            structured_output,
            structured_output_descriptors,
            exposure_policy,
        )
    return replace(self, content=content, structured_output=structured_output)

ModelSelection(provider=None, model=None, profile=None, metadata=dict()) dataclass

Provider/model selector carried by one Agent run.

A service may let a user choose OpenAI, Anthropic, Vertex, OpenRouter, vLLM, or another provider per run. The selector is intentionally provider-neutral: concrete adapters or routing models decide which values they accept.

__post_init__()

Reject blank selector fields before they reach provider adapters.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def __post_init__(self) -> None:
    """Reject blank selector fields before they reach provider adapters."""
    for label, value in (
        ("provider", self.provider),
        ("model", self.model),
        ("profile", self.profile),
    ):
        if value is not None and not value.strip():
            raise AgentDefinitionError(f"Model selection {label} cannot be blank")

ModelStreamEvent(kind, token_delta=None, message_delta=None, reasoning_delta=None, tool_call=None, tool_call_args_delta=None, structured_output=None, error=None, usage=None, metadata=dict()) dataclass

Provider-neutral model streaming event.

token_delta carries the generic streamed token channel. message_delta and reasoning_delta distinguish assistant-facing text from model reasoning so callers can route or suppress reasoning independently. tool_call_args_delta carries incremental tool-call argument text framed by TOOL_CALL_START and TOOL_CALL_END boundary events that reference the same tool_call.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive streaming payloads guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelStreamEvent":
    """Return a copy with sensitive streaming payloads guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    text_channels: dict[str, str | None] = {
        "token_delta": self.token_delta,
        "message_delta": self.message_delta,
        "reasoning_delta": self.reasoning_delta,
        "tool_call_args_delta": self.tool_call_args_delta,
    }
    guarded_text = {
        field_name: self._guard_text_delta(
            field_name, value, sensitive_fields, exposure_policy
        )
        for field_name, value in text_channels.items()
        if value is not None
    }
    text_paths = (
        (),
        *((field_name,) for field_name in text_channels),
    )
    structured_output = self.structured_output
    structured_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path not in text_paths
    )
    if structured_descriptors:
        structured_output = guard_json_value(
            structured_output,
            structured_descriptors,
            exposure_policy,
        )
    tool_call = self.tool_call
    if tool_call is not None:
        tool_call = tool_call.guarded(sensitive_fields, exposure_policy)
    return replace(
        self,
        structured_output=structured_output,
        tool_call=tool_call,
        **guarded_text,
    )

ModelStreamEventKind

Bases: StrEnum

Provider-neutral streaming event kinds emitted by a model adapter.

ModelToolCall(name, arguments, call_id=None, metadata=dict()) dataclass

Tool invocation candidate emitted by a model adapter.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive argument values deterministically guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelToolCall":
    """Return a copy with sensitive argument values deterministically guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    guarded_arguments = guard_json_value(
        self.arguments,
        sensitive_fields,
        exposure_policy,
    )
    if not isinstance(guarded_arguments, Mapping):
        guarded_arguments = {}
    return replace(self, arguments=guarded_arguments)

ModelToolChoice

Bases: StrEnum

Provider-neutral tool calling strategy requested from a model adapter.

ModelToolSpec(name, parameters, description=None, metadata=dict()) dataclass

LLM-facing tool descriptor normalized by agent tooling.

ModelUsage(input_tokens=None, output_tokens=None, total_tokens=None) dataclass

Token accounting reported by a model adapter.

SamplingOptions(temperature=None, top_p=None, max_tokens=None) dataclass

Portable model sampling options.

StreamingOptions(include_usage=True, include_progress=True) dataclass

Portable model streaming options.

StructuredOutputSpec(constraint, output_type_name=None) dataclass

Structured output contract requested from a model adapter.

ToolCallingSpec(tools, choice=ModelToolChoice.AUTO) dataclass

Tool calling contract requested from a model adapter.

AgentSignal(id, agent_state_id, kind, payload=dict(), created_at=None) dataclass

Inbound stimulus appended for an agent execution.

AgentSignalKind

Bases: StrEnum

Inbound stimulus kinds that an agent may accept while running.

ApprovalDecision

Bases: StrEnum

Human-in-the-loop approval outcomes.

AgentSignalConsumptionBatch(state_id, poll_point, signals) dataclass

Signals consumed during one non-blocking poll.

consumed_count property

Return the number of signals consumed by this poll.

AgentSignalPollPoint

Bases: StrEnum

Places where orchestration may poll durable signals without waiting.

ContextExposurePolicy(include_pii_values=False, include_sensitive_values=False, include_sensitive_schema_metadata=False, include_sensitive_context_metadata=False) dataclass

Policy for LLM-facing context and schema metadata exposure.

can_expose_value(field)

Return whether a sensitive value may cross the model boundary.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def can_expose_value(self, field: "SensitiveField | SecretField") -> bool:
    """Return whether a sensitive value may cross the model boundary."""
    if isinstance(field, SecretField):
        return False
    if field.sensitivity == DataSensitivity.PII:
        return self.include_pii_values
    if field.sensitivity in (DataSensitivity.CONFIDENTIAL, DataSensitivity.SECRET):
        return self.include_sensitive_values
    return True

CredentialRef(id, provider=None) dataclass

Reference to a credential outside LLM-facing context.

__post_init__()

Reject blank credential references.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank credential references."""
    _require_non_blank(self.id, "Credential reference id")
    if self.provider is not None:
        _require_non_blank(self.provider, "Credential provider")

DataSensitivity

Bases: StrEnum

Canonical sensitivity classes carried by descriptors.

EvidenceExposurePolicy(include_pii_values=False, include_sensitive_values=False, include_sensitive_metadata=True) dataclass

Policy for evidence payload exposure before append-only capture.

can_expose_value(field)

Return whether a sensitive value may be stored in evidence payloads.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def can_expose_value(self, field: "SensitiveField | SecretField") -> bool:
    """Return whether a sensitive value may be stored in evidence payloads."""
    if isinstance(field, SecretField):
        return False
    if field.sensitivity == DataSensitivity.PII:
        return self.include_pii_values
    if field.sensitivity in (DataSensitivity.CONFIDENTIAL, DataSensitivity.SECRET):
        return self.include_sensitive_values
    return True

MaskingPolicy

Bases: StrEnum

Deterministic text masking strategies for sensitive values.

PII

Bases: StrEnum

PII categories that can be declared with typing.Annotated.

RedactionPolicy

Bases: StrEnum

Boundary action used when a value must not be exposed.

SecretField(redaction=RedactionPolicy.REFERENCE_ONLY, label=None, metadata=dict()) dataclass

typing.Annotated metadata for values that must never be model text.

sensitivity property

Return the normalized sensitivity class.

__post_init__()

Reject blank secret field labels.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank secret field labels."""
    if self.label is not None:
        _require_non_blank(self.label, "Secret field label")

guard_text(value)

Return deterministic replacement text for a secret value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def guard_text(self, value: str) -> str:
    """Return deterministic replacement text for a secret value."""
    return SECRET_VALUE

to_metadata()

Serialize marker metadata without including the secret value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_metadata(self) -> JsonObject:
    """Serialize marker metadata without including the secret value."""
    payload: dict[str, JsonValue] = {
        "kind": "secret",
        "sensitivity": DataSensitivity.SECRET.value,
        "redaction": self.redaction.value,
    }
    if self.label is not None:
        payload["label"] = self.label
    if self.metadata:
        payload["metadata"] = self.metadata
    return payload

SecretRef(id, credential=None) dataclass

Opaque reference to a secret value stored outside model context.

__post_init__()

Reject blank secret references.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank secret references."""
    _require_non_blank(self.id, "Secret reference id")

SensitiveField(category, masking=MaskingPolicy.REDACT, redaction=RedactionPolicy.REDACT, label=None, metadata=dict()) dataclass

typing.Annotated metadata for deterministic sensitive-field handling.

sensitivity property

Return the normalized sensitivity class.

category_name property

Return the stable public category name.

__post_init__()

Reject blank sensitive field labels.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank sensitive field labels."""
    if self.label is not None:
        _require_non_blank(self.label, "Sensitive field label")

guard_text(value)

Return deterministic model/evidence-safe text for this field.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def guard_text(self, value: str) -> str:
    """Return deterministic model/evidence-safe text for this field."""
    return _mask_text(value, self.masking, self.category_name)

to_metadata()

Serialize marker metadata without including the sensitive value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_metadata(self) -> JsonObject:
    """Serialize marker metadata without including the sensitive value."""
    payload: dict[str, JsonValue] = {
        "kind": "sensitive",
        "category": self.category_name,
        "sensitivity": self.sensitivity.value,
        "masking": self.masking.value,
        "redaction": self.redaction.value,
    }
    if self.label is not None:
        payload["label"] = self.label
    if self.metadata:
        payload["metadata"] = self.metadata
    return payload

SensitiveFieldDescriptor(path, field) dataclass

Path-bound sensitive metadata extracted from Annotated types.

to_metadata()

Serialize descriptor metadata without leaking the field value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_metadata(self) -> JsonObject:
    """Serialize descriptor metadata without leaking the field value."""
    return {
        "path": list(self.path),
        "field": self.field.to_metadata(),
    }

StreamingGuardFailureMode

Bases: StrEnum

Final audit behavior when a streaming guard missed a raw candidate.

StreamingRedactionAudit(status, detected_count, redacted_count, missed_matches, buffer_size, emitted_char_count, original_char_count) dataclass

Final aggregate audit for a bounded streaming redaction session.

missed_count property

Return the number of raw candidates still present after streaming.

to_evidence_payload()

Serialize audit evidence without raw streamed content.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_evidence_payload(self) -> JsonObject:
    """Serialize audit evidence without raw streamed content."""
    return {
        "kind": "streaming_redaction_audit",
        "status": self.status.value,
        "detected_count": self.detected_count,
        "redacted_count": self.redacted_count,
        "missed_count": self.missed_count,
        "missed_matches": tuple(
            match.to_payload() for match in self.missed_matches
        ),
        "buffer_size": self.buffer_size,
        "emitted_char_count": self.emitted_char_count,
        "original_char_count": self.original_char_count,
    }

to_error_payload()

Serialize a typed error payload for stream consumers.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_error_payload(self) -> JsonObject:
    """Serialize a typed error payload for stream consumers."""
    return {
        "code": "streaming_redaction_audit_failed",
        "message": "Streaming redaction final audit detected unmasked candidates",
        "retryable": False,
        "metadata": self.to_evidence_payload(),
    }

StreamingRedactionAuditStatus

Bases: StrEnum

Final aggregate streaming redaction audit status.

StreamingRedactionMatch(pattern_name, start, end, metadata=dict()) dataclass

Sanitized final-audit match location for a missed redaction candidate.

to_payload()

Serialize a match without the sensitive value itself.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_payload(self) -> JsonObject:
    """Serialize a match without the sensitive value itself."""
    payload: dict[str, JsonValue] = {
        "pattern_name": self.pattern_name,
        "start": self.start,
        "end": self.end,
    }
    if self.metadata:
        payload["metadata"] = self.metadata
    return payload

StreamingRedactionPolicy(patterns, buffer_size=64, emit_chunk_size=None, failure_mode=StreamingGuardFailureMode.RAISE) dataclass

Bounded buffering policy balancing stream latency and redaction correctness.

__post_init__()

Reject policies that would make the guard unbounded or silent.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject policies that would make the guard unbounded or silent."""
    if self.buffer_size <= 0:
        raise AgentDefinitionError(
            "Streaming redaction buffer size must be positive"
        )
    if self.emit_chunk_size is not None and self.emit_chunk_size <= 0:
        raise AgentDefinitionError(
            "Streaming redaction emit chunk size must be positive"
        )
    if len(self.patterns) == 0:
        raise AgentDefinitionError("Streaming redaction patterns cannot be empty")

StreamingRedactionResult(chunks, audit=None, error=None) dataclass

Output produced by a bounded streaming redaction step.

StreamingRedactionSession(policy)

Stateful bounded redactor for model token streams.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __init__(self, policy: StreamingRedactionPolicy) -> None:
    self._policy = policy
    self._patterns = tuple(policy.patterns)
    self._buffer = ""
    self._original_chunks: list[str] = []
    self._emitted_chunks: list[str] = []
    self._redacted_count = 0
    self._closed = False

push(chunk)

Redact one token chunk and emit only the safe bounded prefix.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def push(self, chunk: str) -> StreamingRedactionResult:
    """Redact one token chunk and emit only the safe bounded prefix."""
    self._ensure_open()
    if chunk == "":
        return StreamingRedactionResult(chunks=())
    self._original_chunks.append(chunk)
    self._buffer = self._redact_text(f"{self._buffer}{chunk}")
    chunks = self._emit_safe_prefix()
    return StreamingRedactionResult(chunks=chunks)

finish()

Flush the remaining buffer and run the mandatory final audit.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def finish(self) -> StreamingRedactionResult:
    """Flush the remaining buffer and run the mandatory final audit."""
    self._ensure_open()
    self._closed = True
    redacted_buffer = self._redact_text(self._buffer)
    chunks = self._split_for_emit(redacted_buffer)
    self._buffer = ""
    self._emitted_chunks.extend(chunks)
    audit = self._audit()
    if audit.status == StreamingRedactionAuditStatus.FAILED:
        if self._policy.failure_mode == StreamingGuardFailureMode.RAISE:
            raise AgentOutputGuardError(
                "Streaming redaction final audit detected unmasked candidates"
            )
        return StreamingRedactionResult(
            chunks=chunks,
            audit=audit,
            error=audit.to_error_payload(),
        )
    return StreamingRedactionResult(chunks=chunks, audit=audit)

StreamingSensitivePattern(name, pattern, replacement=REDACTED_VALUE, metadata=dict()) dataclass

Caller-supplied deterministic pattern used by streaming redaction.

__post_init__()

Reject blank names and invalid pattern syntax before streaming starts.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank names and invalid pattern syntax before streaming starts."""
    _require_non_blank(self.name, "Streaming sensitive pattern name")
    _require_non_blank(self.pattern, "Streaming sensitive pattern")
    try:
        re.compile(self.pattern)
    except re.error as e:
        raise AgentDefinitionError("Streaming sensitive pattern is invalid") from e

redact(value)

Return redacted text and the number of replacements applied.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def redact(self, value: str) -> tuple[str, int]:
    """Return redacted text and the number of replacements applied."""
    compiled = re.compile(self.pattern)
    return compiled.subn(self.replacement, value)

find_matches(value)

Return sanitized match locations without exposing raw text.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def find_matches(self, value: str) -> tuple["StreamingRedactionMatch", ...]:
    """Return sanitized match locations without exposing raw text."""
    compiled = re.compile(self.pattern)
    return tuple(
        StreamingRedactionMatch(
            pattern_name=self.name,
            start=match.start(),
            end=match.end(),
            metadata=self.metadata,
        )
        for match in compiled.finditer(value)
    )

AgentState(id, agent_type, status, transition=None, reason=None, current_activity=None, input_ref=None, output_ref=None, pending_signal_count=0, last_event_cursor=None, recovery_marker=None, metadata=dict(), created_at=None, updated_at=None) dataclass

Materialized state for a long-running agent execution.

__post_init__()

Reject state snapshots that cannot represent a real queue count.

Source code in core/spakky-agent/src/spakky/agent/state.py

def __post_init__(self) -> None:
    """Reject state snapshots that cannot represent a real queue count."""
    if self.pending_signal_count < 0:
        raise AgentDefinitionError("Agent pending signal count cannot be negative")

AgentStateReason

Bases: StrEnum

Structured reason that refines an externally observable lifecycle state.

AgentStateTransition

Bases: StrEnum

State transition vocabulary accepted by durable agent orchestration.

AgentStatus

Bases: StrEnum

Externally observable lifecycle states for an agent execution.

AgentToolCatalog(descriptors=()) dataclass

Deterministic catalog of descriptors discovered from an Agent class.

__post_init__()

Reject duplicate identity or schema names before model lookup.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject duplicate identity or schema names before model lookup."""
    identity_keys: set[str] = set()
    schema_names: set[str] = set()
    for descriptor in self.descriptors:
        if descriptor.identity.key in identity_keys:
            raise AgentDefinitionError("Agent tool identity must be unique")
        if descriptor.schema.name in schema_names:
            raise AgentDefinitionError("Agent tool schema name must be unique")
        identity_keys.add(descriptor.identity.key)
        schema_names.add(descriptor.schema.name)

by_identity(identity)

Lookup a descriptor by typed identity.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def by_identity(self, identity: AgentToolIdentity) -> AgentToolDescriptor:
    """Lookup a descriptor by typed identity."""
    for descriptor in self.descriptors:
        if descriptor.identity == identity:
            return descriptor
    raise AgentDefinitionError("Agent tool identity is not registered")

by_schema_name(schema_name)

Lookup a descriptor by model-facing schema name.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def by_schema_name(self, schema_name: str) -> AgentToolDescriptor:
    """Lookup a descriptor by model-facing schema name."""
    _require_non_blank(schema_name, "Agent tool schema name")
    for descriptor in self.descriptors:
        if descriptor.schema.name == schema_name:
            return descriptor
    raise AgentDefinitionError("Agent tool schema name is not registered")

AgentToolApprovalContext(prompt=None, action_ref=None, metadata=dict()) dataclass

Invocation-specific approval display and correlation metadata.

__post_init__()

Reject blank override strings before they enter approval state.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject blank override strings before they enter approval state."""
    if self.prompt is not None:
        _require_non_blank(self.prompt, "Agent tool approval prompt")
    if self.action_ref is not None:
        _require_non_blank(self.action_ref, "Agent tool approval action ref")

AgentToolBoundInvocation(args=(), kwargs=dict()) dataclass

Python-call-ready tool invocation arguments.

AgentToolDefinition(name, schema_name, description=None, metadata=AgentToolMetadata()) dataclass

Method-level metadata attached by @agent_tool before owner discovery.

__post_init__()

Reject definitions that would make catalog lookup ambiguous.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject definitions that would make catalog lookup ambiguous."""
    _require_non_blank(self.name, "Agent tool name")
    _require_non_blank(self.schema_name, "Agent tool schema name")
    if self.description is not None and not self.description.strip():
        raise AgentDefinitionError("Agent tool description cannot be blank")

AgentToolDescriptor(identity, owner, callable, schema, description=None, metadata=AgentToolMetadata()) dataclass

Discovered tool descriptor bound to an owner class and callable.

name property

Return the descriptor-local tool name.

bind_invocation(payload)

Bind a decoded model payload before the tool callable is invoked.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def bind_invocation(self, payload: JsonObject) -> "AgentToolBoundInvocation":
    """Bind a decoded model payload before the tool callable is invoked."""
    return bind_agent_tool_invocation(self.callable, payload)

approval_context(payload)

Return invocation-specific approval context for this payload.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def approval_context(self, payload: JsonObject) -> AgentToolApprovalContext:
    """Return invocation-specific approval context for this payload."""
    _ = payload
    return AgentToolApprovalContext()

AgentToolIdentity(owner_module, owner_qualname, name) dataclass

Descriptor identity independent from free-form model output text.

key property

Return a stable key for maps, logs, and evidence metadata.

__post_init__()

Reject identity parts that cannot be serialized or matched.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject identity parts that cannot be serialized or matched."""
    _require_non_blank(self.owner_module, "Agent tool owner module")
    _require_non_blank(self.owner_qualname, "Agent tool owner qualname")
    _require_non_blank(self.name, "Agent tool name")

AgentToolMetadata(permissions=(), effects=ToolEffects(), idempotency=Idempotency.UNKNOWN, data_access=DataAccess.NONE, externality=Externality.LOCAL, timeout=TimeoutPolicy(), result_budget=ResultBudget(), evidence=EvidenceCapture.NONE, approval=ToolApprovalRequirement.DERIVED) dataclass

Typed approval, idempotency, and evidence metadata for a descriptor.

risk property

Return derived risk axes without storing risk as source metadata.

resume property

Return resume metadata derived from stored idempotency.

requires_approval_candidate property

Return whether this tool is a HITL approval candidate.

AgentToolRuntimeContext(state_id, conversation_id, call_id, tool_name) dataclass

Runtime correlation data injected into framework-owned tool callables.

__post_init__()

Reject runtime context that cannot link events and evidence.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject runtime context that cannot link events and evidence."""
    _require_non_blank(self.state_id, "Agent tool runtime state id")
    _require_non_blank(self.conversation_id, "Agent tool runtime conversation id")
    _require_non_blank(self.call_id, "Agent tool runtime call id")
    _require_non_blank(self.tool_name, "Agent tool runtime tool name")

AgentToolSchemaHandle(name, input_schema_name, output_schema_name, input_schema=dict(), output_schema=dict(), input_sensitive_fields=(), output_sensitive_fields=()) dataclass

Stable schema names and generated JSON schemas owned by a descriptor.

__post_init__()

Reject blank schema handles.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject blank schema handles."""
    _require_non_blank(self.name, "Agent tool schema name")
    _require_non_blank(self.input_schema_name, "Agent tool input schema name")
    _require_non_blank(self.output_schema_name, "Agent tool output schema name")

input_schema_for(policy=None)

Return model-facing input schema under the requested exposure policy.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def input_schema_for(
    self,
    policy: ContextExposurePolicy | None = None,
) -> JsonObject:
    """Return model-facing input schema under the requested exposure policy."""
    return schema_with_sensitive_metadata(
        self.input_schema,
        self.input_sensitive_fields,
        policy or ContextExposurePolicy(),
    )

output_schema_for(policy=None)

Return model-facing output schema under the requested exposure policy.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def output_schema_for(
    self,
    policy: ContextExposurePolicy | None = None,
) -> JsonObject:
    """Return model-facing output schema under the requested exposure policy."""
    return schema_with_sensitive_metadata(
        self.output_schema,
        self.output_sensitive_fields,
        policy or ContextExposurePolicy(),
    )

DataAccess

Bases: StrEnum

Data access level declared by a tool.

EvidenceCapture

Bases: StrEnum

Evidence capture strategy for tool results.

Externality

Bases: StrEnum

External side-effect boundary declared by a tool.

Idempotency

Bases: StrEnum

Action idempotency declared by a tool.

ResultBudget(max_bytes=None) dataclass

Optional result-size budget for model-facing tool output.

__post_init__()

Reject result budgets that cannot constrain output.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject result budgets that cannot constrain output."""
    if self.max_bytes is not None and self.max_bytes <= 0:
        raise AgentDefinitionError("Agent tool result budget must be positive")

TimeoutPolicy(seconds=None) dataclass

Optional timeout boundary for a tool invocation.

__post_init__()

Reject non-positive timeout policies at definition time.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject non-positive timeout policies at definition time."""
    if self.seconds is not None and self.seconds <= 0:
        raise AgentDefinitionError("Agent tool timeout must be positive")

ToolApprovalRequirement

Bases: StrEnum

Human approval requirement at the tool boundary.

ToolEffects(data_access=DataAccess.NONE, externality=Externality.LOCAL, destructive=False, network=False) dataclass

Typed effect metadata used to derive display risk outside core.

read_only() classmethod

Declare a local read-only tool.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def read_only(cls) -> "ToolEffects":
    """Declare a local read-only tool."""
    return cls(data_access=DataAccess.READ, externality=Externality.LOCAL)

write_state() classmethod

Declare a local state-writing tool.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def write_state(cls) -> "ToolEffects":
    """Declare a local state-writing tool."""
    return cls(data_access=DataAccess.WRITE, externality=Externality.LOCAL)

external_side_effect() classmethod

Declare a tool that crosses an external side-effect boundary.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def external_side_effect(cls) -> "ToolEffects":
    """Declare a tool that crosses an external side-effect boundary."""
    return cls(
        data_access=DataAccess.READ_WRITE,
        externality=Externality.EXTERNAL,
        network=True,
    )

destructive_action() classmethod

Declare a tool that may irreversibly mutate local or external state.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def destructive_action(cls) -> "ToolEffects":
    """Declare a tool that may irreversibly mutate local or external state."""
    return cls(
        data_access=DataAccess.WRITE,
        externality=Externality.EXTERNAL,
        destructive=True,
    )

ToolPermission(name) dataclass

Typed permission marker attached to a tool descriptor.

__post_init__()

Reject permission names that cannot be matched deterministically.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject permission names that cannot be matched deterministically."""
    if not self.name.strip():
        raise AgentDefinitionError("Agent tool permission name cannot be blank")

ToolResumeAction

Bases: StrEnum

Resume action allowed by stored tool idempotency metadata.

ToolResumeMetadata(idempotency=Idempotency.UNKNOWN) dataclass

Stored idempotency metadata used when resuming an incomplete action.

from_metadata(metadata) classmethod

Build resume metadata from a tool descriptor metadata object.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def from_metadata(cls, metadata: "AgentToolMetadata") -> "ToolResumeMetadata":
    """Build resume metadata from a tool descriptor metadata object."""
    return cls(idempotency=metadata.idempotency)

action_for_completed_boundary()

Return the resume action for an already completed action boundary.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def action_for_completed_boundary(self) -> ToolResumeAction:
    """Return the resume action for an already completed action boundary."""
    return ToolResumeAction.SKIP_COMPLETED

action_for_incomplete_boundary()

Return the resume action for an incomplete action boundary.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def action_for_incomplete_boundary(self) -> ToolResumeAction:
    """Return the resume action for an incomplete action boundary."""
    if self.idempotency in (
        Idempotency.IDEMPOTENT,
        Idempotency.CONDITIONALLY_IDEMPOTENT,
    ):
        return ToolResumeAction.RETRY
    return ToolResumeAction.REQUIRE_APPROVAL

ToolRisk(axes=()) dataclass

Derived typed risk axes for policy and evidence annotations.

requires_approval_candidate property

Return whether the risk is strong enough to suggest HITL approval.

__post_init__()

Reject duplicate axes so risk comparisons stay deterministic.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject duplicate axes so risk comparisons stay deterministic."""
    seen: set[ToolRiskAxis] = set()
    for axis in self.axes:
        if axis in seen:
            raise AgentDefinitionError("Agent tool risk axes must be unique")
        seen.add(axis)

from_metadata(metadata) classmethod

Derive risk axes from source tool metadata.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def from_metadata(cls, metadata: AgentToolMetadata) -> "ToolRisk":
    """Derive risk axes from source tool metadata."""
    axes: list[ToolRiskAxis] = []
    if metadata.data_access in (DataAccess.READ, DataAccess.READ_WRITE):
        axes.append(ToolRiskAxis.READ)
    if metadata.data_access in (DataAccess.WRITE, DataAccess.READ_WRITE):
        axes.append(ToolRiskAxis.WRITE)
    if metadata.data_access in (DataAccess.WRITE, DataAccess.READ_WRITE):
        axes.append(ToolRiskAxis.SIDE_EFFECT)
    if metadata.externality in (Externality.EXTERNAL, Externality.UNKNOWN):
        axes.append(ToolRiskAxis.SIDE_EFFECT)
    if metadata.effects.destructive:
        axes.append(ToolRiskAxis.DESTRUCTIVE)
    if metadata.effects.network or metadata.externality == Externality.EXTERNAL:
        axes.append(ToolRiskAxis.NETWORK)
    return cls(axes=_deduplicate_axes(axes))

includes(axis)

Return whether this risk contains the requested axis.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def includes(self, axis: ToolRiskAxis) -> bool:
    """Return whether this risk contains the requested axis."""
    return axis in self.axes

ToolRiskAxis

Bases: StrEnum

Derived risk axes exposed for policy and UI decisions.

AgentYield(kind, payload) dataclass

Typed stream item returned from an agent execute generator.

AgentYieldKind

Bases: StrEnum

Canonical public vocabulary yielded by agent execution.

Approval(id, prompt, allowed_decisions, metadata) dataclass

Approval request surfaced to the inbound adapter.

Cancel(reason=None, requested_by=None, metadata=dict()) dataclass

Cancellation acknowledgement surfaced to the caller.

Error(code, message, retryable=False, metadata=dict()) dataclass

Recoverable or terminal execution error surfaced to the caller.

Evidence(evidence, metadata=dict()) dataclass

Evidence item surfaced to the inbound adapter.

Final(output, metadata) dataclass

Final output carried by a generator stream.

Progress(message, current_step=None, metadata=dict()) dataclass

Agent progress update intended for direct inbound adapter consumption.

Token(text, metadata=dict()) dataclass

Incremental model token intended for streaming clients.

Tool(name, call_id=None, arguments=dict(), result=None, metadata=dict()) dataclass

Tool call or tool result surfaced by agent execution.

begin_agent_cancellation(state, signal)

Materialize receipt of a CANCEL signal as CANCELLING state.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def begin_agent_cancellation(
    state: AgentState,
    signal: AgentSignal,
) -> AgentState:
    """Materialize receipt of a CANCEL signal as CANCELLING state."""
    _require_cancel_signal(signal)
    _require_signal_matches_state(state, signal)
    return replace(
        state,
        status=AgentStatus.CANCELLING,
        transition=AgentStateTransition.CANCELLING,
        reason=AgentStateReason.CANCELLATION_REQUESTED,
        metadata={
            **state.metadata,
            "cancellation_signal_id": signal.id,
            "cancellation_reason": _optional_string(signal.payload, "reason"),
            "cancellation_requested_by": _optional_string(
                signal.payload,
                "requested_by",
            ),
        },
    )

complete_agent_cancellation(state, report)

Resolve CANCELLING state into CANCELLED or FAILED cleanup outcome.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def complete_agent_cancellation(
    state: AgentState,
    report: AgentCancellationCleanupReport,
) -> AgentState:
    """Resolve CANCELLING state into CANCELLED or FAILED cleanup outcome."""
    if state.id != report.state_id:
        raise AgentDefinitionError("Agent cancellation report state id mismatch")
    metadata = {**state.metadata, "cancellation_cleanup": report.to_payload()}
    if report.cleanup_succeeded:
        return replace(
            state,
            status=AgentStatus.CANCELLED,
            transition=AgentStateTransition.CANCELLED,
            reason=AgentStateReason.CANCELLATION_REQUESTED,
            metadata=metadata,
        )
    return replace(
        state,
        status=AgentStatus.FAILED,
        transition=AgentStateTransition.FAILED,
        reason=AgentStateReason.CANCELLATION_CLEANUP_FAILED,
        metadata=metadata,
    )

run_agent_cancellation_cleanup(*, state, signal, tasks) async

Invoke model stream/tool/delegate cleanup hooks for a CANCEL signal.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

async def run_agent_cancellation_cleanup(
    *,
    state: AgentState,
    signal: AgentSignal,
    tasks: Sequence[AgentCancellationCleanupTask],
) -> AgentCancellationCleanupReport:
    """Invoke model stream/tool/delegate cleanup hooks for a CANCEL signal."""
    _require_cancel_signal(signal)
    outcomes: list[AgentCancellationCleanupResult] = []
    for task in tasks:
        outcomes.append(await task.run(state=state, signal=signal))
    return AgentCancellationCleanupReport(
        state_id=state.id,
        signal_id=signal.id,
        outcomes=tuple(outcomes),
    )

materialize_agent_approval_decision_state(current, outcome)

Apply a typed approval decision outcome to an existing state snapshot.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def materialize_agent_approval_decision_state(
    current: AgentState,
    outcome: AgentApprovalDecisionOutcome,
) -> AgentState:
    """Apply a typed approval decision outcome to an existing state snapshot."""
    return AgentState(
        id=current.id,
        agent_type=current.agent_type,
        status=outcome.status,
        transition=outcome.transition,
        reason=outcome.reason,
        current_activity=_activity_for_decision(outcome),
        input_ref=current.input_ref,
        output_ref=current.output_ref,
        pending_signal_count=current.pending_signal_count,
        last_event_cursor=current.last_event_cursor,
        recovery_marker=current.recovery_marker,
        metadata={
            **current.metadata,
            "approval_decision": {
                "request_id": outcome.request_id,
                "decision": outcome.decision.value,
                "modified_payload": outcome.modified_payload,
                "comment": outcome.comment,
            },
        },
        created_at=current.created_at,
        updated_at=current.updated_at,
    )

parse_agent_approval_decision_signal(signal, *, request=None)

Parse an approval decision signal into a typed workflow outcome.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def parse_agent_approval_decision_signal(
    signal: AgentSignal,
    *,
    request: AgentApprovalRequest | None = None,
) -> AgentApprovalDecisionOutcome:
    """Parse an approval decision signal into a typed workflow outcome."""
    if signal.kind != AgentSignalKind.APPROVAL_DECISION:
        raise AgentDefinitionError(
            "Agent approval signal kind must be approval_decision"
        )

    request_id = _require_payload_text(signal.payload, "request_id")
    if request is not None and request_id != request.id:
        raise AgentDefinitionError("Agent approval signal request id does not match")
    if request is not None and signal.agent_state_id != request.agent_state_id:
        raise AgentDefinitionError("Agent approval signal state id does not match")

    decision = _parse_approval_decision(
        _require_payload_text(signal.payload, "decision")
    )
    modified_payload = _payload_object(signal.payload.get("modified_payload", {}))
    comment = _optional_payload_text(signal.payload, "comment")
    status, transition, reason = _decision_state_target(decision)

    return AgentApprovalDecisionOutcome(
        request_id=request_id,
        decision=decision,
        status=status,
        transition=transition,
        reason=reason,
        modified_payload=modified_payload,
        comment=comment,
    )

plan_agent_tool_approval(*, descriptor, approval_id, agent_state_id, agent_type, prompt=None, action_ref=None, call_id=None, metadata=None)

Plan whether a tool invocation should proceed or wait for approval.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def plan_agent_tool_approval(
    *,
    descriptor: AgentToolDescriptor,
    approval_id: str,
    agent_state_id: str,
    agent_type: str,
    prompt: str | None = None,
    action_ref: str | None = None,
    call_id: str | None = None,
    metadata: JsonObject | None = None,
) -> AgentApprovalPlan:
    """Plan whether a tool invocation should proceed or wait for approval."""
    if not descriptor.metadata.requires_approval_candidate:
        return AgentApprovalPlan(action=AgentApprovalPlanAction.PROCEED)

    request = AgentApprovalRequest.from_tool_descriptor(
        approval_id=approval_id,
        agent_state_id=agent_state_id,
        descriptor=descriptor,
        prompt=prompt,
        action_ref=action_ref,
        call_id=call_id,
        metadata=metadata,
    )
    return AgentApprovalPlan(
        action=AgentApprovalPlanAction.WAIT_FOR_APPROVAL,
        request=request,
        state=request.to_state(agent_type=agent_type),
        yield_item=request.to_yield(),
    )

plan_agent_resume(state, evidence, signals=())

Restore the next resume action using only persisted state and evidence.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

def plan_agent_resume(
    state: AgentState,
    evidence: Sequence[AgentEvidence],
    signals: Sequence[AgentSignal] = (),
) -> AgentResumePlan:
    """Restore the next resume action using only persisted state and evidence."""
    persisted_signals = tuple(signals)
    if state.status in (
        AgentStatus.COMPLETED,
        AgentStatus.FAILED,
        AgentStatus.CANCELLED,
    ):
        return AgentResumePlan(
            state=state,
            action=AgentResumeAction.NOT_RESUMABLE,
            signals=persisted_signals,
        )

    boundary = _last_action_boundary(evidence)
    if boundary is None:
        return AgentResumePlan(
            state=state,
            action=AgentResumeAction.START,
            signals=persisted_signals,
        )

    if boundary.stage is AgentActionBoundaryStage.AFTER:
        return AgentResumePlan(
            state=replace(state, recovery_marker=boundary.action_id),
            action=AgentResumeAction.SKIP_COMPLETED,
            boundary=boundary,
            signals=persisted_signals,
        )

    if boundary.action_kind is AgentActionKind.APPROVAL_WAIT:
        return AgentResumePlan(
            state=_interrupt_for_human_recovery(state, boundary),
            action=AgentResumeAction.REQUIRE_HITL,
            boundary=boundary,
            signals=persisted_signals,
        )

    resume_action = ToolResumeMetadata(
        idempotency=boundary.idempotency,
    ).action_for_incomplete_boundary()
    if resume_action is ToolResumeAction.RETRY:
        return AgentResumePlan(
            state=replace(state, recovery_marker=boundary.action_id),
            action=AgentResumeAction.RETRY,
            boundary=boundary,
            signals=persisted_signals,
        )
    return AgentResumePlan(
        state=_interrupt_for_human_recovery(state, boundary),
        action=AgentResumeAction.REQUIRE_HITL,
        boundary=boundary,
        signals=persisted_signals,
    )

discover_agent_signal_hooks(owner)

Discover @on_signal methods in deterministic class-definition order.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def discover_agent_signal_hooks(owner: type[object]) -> AgentSignalHookCatalog:
    """Discover ``@on_signal`` methods in deterministic class-definition order."""
    descriptors: list[AgentSignalHookDescriptor] = []
    resolved_member_names: set[str] = set()
    for declaring_owner in owner.__mro__:
        if declaring_owner is object:
            continue
        for member_name, member in vars(declaring_owner).items():
            if member_name in resolved_member_names:
                continue
            resolved_member_names.add(member_name)
            if not isinstance(member, FunctionType):
                continue
            definition = _get_signal_hook_definition(member)
            if definition is None:
                continue
            _validate_hook_contract(member)
            descriptors.append(
                AgentSignalHookDescriptor(
                    identity=AgentSignalHookIdentity(
                        owner_module=declaring_owner.__module__,
                        owner_qualname=declaring_owner.__qualname__,
                        member_name=member_name,
                    ),
                    owner=declaring_owner,
                    callable=member,
                    kind=definition.kind,
                ),
            )
    ordered = tuple(
        sorted(descriptors, key=lambda descriptor: descriptor.identity.key),
    )
    return AgentSignalHookCatalog(descriptors=ordered)

on_signal(kind)

Declare a method that reacts to one inbound signal kind.

The decorated method must be an async generator yielding AgentYield items and accepting exactly one signal: AgentSignal argument besides self. The runner invokes it when it consumes a signal of kind at a poll point, and forwards every yielded item into the public stream.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def on_signal(kind: AgentSignalKind) -> Callable[[FunctionType], FunctionType]:
    """Declare a method that reacts to one inbound signal kind.

    The decorated method must be an async generator yielding ``AgentYield``
    items and accepting exactly one ``signal: AgentSignal`` argument besides
    ``self``. The runner invokes it when it consumes a signal of ``kind`` at a
    poll point, and forwards every yielded item into the public stream.
    """

    def decorate(function: FunctionType) -> FunctionType:
        function.__dict__[AGENT_SIGNAL_HOOK_DEFINITION_KEY] = AgentSignalHookDefinition(
            kind=kind,
        )
        return function

    return decorate

consume_pending_agent_signals(repository, state_id, *, poll_point=AgentSignalPollPoint.SAFE_BOUNDARY, accepted_signals=None, max_signals=None)

Consume currently pending signals without sleeping or waiting for new input.

The repository remains responsible for durable queue ordering. This helper consumes only the eligible prefix of the pending queue so later signals never overtake an earlier unaccepted signal.

Source code in core/spakky-agent/src/spakky/agent/signal_consumption.py

def consume_pending_agent_signals(
    repository: IAgentSignalRepository,
    state_id: str,
    *,
    poll_point: AgentSignalPollPoint = AgentSignalPollPoint.SAFE_BOUNDARY,
    accepted_signals: Sequence[AgentSignalKind] | None = None,
    max_signals: int | None = None,
) -> AgentSignalConsumptionBatch:
    """Consume currently pending signals without sleeping or waiting for new input.

    The repository remains responsible for durable queue ordering. This helper
    consumes only the eligible prefix of the pending queue so later signals never
    overtake an earlier unaccepted signal.
    """
    if not state_id.strip():
        raise AgentDefinitionError("Agent signal state id cannot be blank")
    if max_signals is not None and max_signals <= 0:
        raise AgentDefinitionError("Agent signal max_signals must be positive")

    accepted = frozenset(accepted_signals) if accepted_signals is not None else None
    pending = tuple(repository.list_pending(state_id))
    selected = _select_consumable_prefix(
        pending,
        accepted_signals=accepted,
        max_signals=max_signals,
    )
    consumed = tuple(repository.mark_consumed(signal.id) for signal in selected)

    return AgentSignalConsumptionBatch(
        state_id=state_id,
        poll_point=poll_point,
        signals=consumed,
    )

agent_tool(*, name=None, schema_name=None, description=None, permissions=(), effects=None, idempotency=Idempotency.UNKNOWN, data_access=None, externality=None, timeout=None, result_budget=None, evidence=EvidenceCapture.NONE, approval=ToolApprovalRequirement.DERIVED)

Attach typed agent-tool metadata to a method object.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def agent_tool(
    *,
    name: str | None = None,
    schema_name: str | None = None,
    description: str | None = None,
    permissions: Sequence[ToolPermission] = (),
    effects: ToolEffects | None = None,
    idempotency: Idempotency = Idempotency.UNKNOWN,
    data_access: DataAccess | None = None,
    externality: Externality | None = None,
    timeout: TimeoutPolicy | None = None,
    result_budget: ResultBudget | None = None,
    evidence: EvidenceCapture = EvidenceCapture.NONE,
    approval: ToolApprovalRequirement = ToolApprovalRequirement.DERIVED,
) -> Callable[[FunctionType], FunctionType]:
    """Attach typed agent-tool metadata to a method object."""

    def decorate(function: FunctionType) -> FunctionType:
        tool_name = _normalize_name(name, function.__name__, "Agent tool name")
        normalized_schema_name = _normalize_name(
            schema_name,
            tool_name,
            "Agent tool schema name",
        )
        tool_effects = effects or ToolEffects()
        metadata = AgentToolMetadata(
            permissions=tuple(permissions),
            effects=tool_effects,
            idempotency=idempotency,
            data_access=data_access or tool_effects.data_access,
            externality=externality or tool_effects.externality,
            timeout=timeout or TimeoutPolicy(),
            result_budget=result_budget or ResultBudget(),
            evidence=evidence,
            approval=approval,
        )
        definition = AgentToolDefinition(
            name=tool_name,
            schema_name=normalized_schema_name,
            description=description,
            metadata=metadata,
        )
        function.__dict__[AGENT_TOOL_DEFINITION_KEY] = definition
        return function

    return decorate

bind_agent_tool_invocation(function, payload)

Bind a structured model payload to a tool function signature.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def bind_agent_tool_invocation(
    function: AgentToolCallable,
    payload: JsonObject,
) -> AgentToolBoundInvocation:
    """Bind a structured model payload to a tool function signature."""
    invocation = _parse_invocation_payload(payload)
    try:
        function_signature = signature(function)
    except (TypeError, ValueError) as e:
        raise AgentToolBindingError(
            "Agent tool callable signature cannot be inspected"
        ) from e
    call_signature = _drop_injected_parameters(function_signature)
    try:
        bound = call_signature.bind(*invocation.args, **invocation.kwargs)
    except TypeError as e:
        raise AgentToolBindingError(
            "Agent tool invocation payload cannot be bound"
        ) from e
    bound.apply_defaults()
    return AgentToolBoundInvocation(
        args=tuple(bound.args),
        kwargs=dict(bound.kwargs),
    )

discover_agent_tools(owner)

Discover @agent_tool methods in deterministic class-definition order.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def discover_agent_tools(owner: type[object]) -> AgentToolCatalog:
    """Discover @agent_tool methods in deterministic class-definition order."""
    descriptors: list[AgentToolDescriptor] = []
    resolved_member_names: set[str] = set()
    for declaring_owner in owner.__mro__:
        if declaring_owner is object:
            continue
        for member_name, member in vars(declaring_owner).items():
            if member_name in resolved_member_names:
                continue
            resolved_member_names.add(member_name)
            function = _unwrap_function(member)
            if function is None:
                continue
            definition = get_agent_tool_definition(function)
            if definition is None:
                continue
            descriptors.append(
                _build_descriptor(declaring_owner, function, definition),
            )
    ordered = tuple(
        sorted(
            descriptors,
            key=lambda descriptor: descriptor.identity.key,
        ),
    )
    return AgentToolCatalog(descriptors=ordered)

실행

Agent execution metadata contracts.

REMOTE_AGENT_CARD_SCHEMES = ('http', 'https') module-attribute

URL schemes accepted for a remote teammate's AgentCard endpoint.

RecoveryStrategy

Bases: StrEnum

Recovery strategy requested by an agent execution contract.

StreamingExposureMode

Bases: StrEnum

Streaming output guard profile exposed by agent execution.

AgentExecutionLimits(timeout_seconds=None) dataclass

Bounded execution limits declared outside infrastructure capabilities.

__post_init__()

Reject limits that would fail later at bootstrap.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject limits that would fail later at bootstrap."""
    if self.timeout_seconds is not None and self.timeout_seconds <= 0:
        raise AgentDefinitionError("Agent timeout limit must be positive")

AgentTeammate(name, pod=None, card_url=None) dataclass

Declared collaborator an agent may delegate to during execution.

A teammate is resolved either in-process by a local @Agent Pod type or remotely by an AgentCard endpoint URL. Exactly one binding is declared; the runtime exposes it as a synthetic teammate delegation tool and dispatches it through the local runner or an injected IAgentDelegate port.

__post_init__()

Reject teammates that cannot resolve to a single delegate target.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject teammates that cannot resolve to a single delegate target."""
    if not self.name.strip():
        raise AgentDefinitionError("Agent teammate name cannot be blank")
    if (self.pod is None) == (self.card_url is None):
        raise AgentDefinitionError(
            "Agent teammate must declare exactly one of a local pod "
            "or a remote AgentCard url"
        )
    if self.pod is not None and not isclass(self.pod):
        raise AgentDefinitionError("Agent teammate pod must be a class")
    if self.card_url is not None:
        parts = urlsplit(self.card_url)
        if parts.scheme not in REMOTE_AGENT_CARD_SCHEMES or not parts.netloc:
            raise AgentDefinitionError(
                "Agent teammate AgentCard url must be an http(s) endpoint"
            )

AgentCompactionPolicy(strategies, trigger_token_threshold) dataclass

Declared compaction chain plus the token threshold that triggers it.

The strategies form an ordered chain of pluggable ICompactionStrategy ports applied in sequence once the running token estimate crosses trigger_token_threshold. The runner threads each strategy's output into the next, so chain order is the compaction order (ADR-0013 §7).

__post_init__()

Reject compaction policies that cannot be enforced consistently.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject compaction policies that cannot be enforced consistently."""
    if not self.strategies:
        raise AgentDefinitionError(
            "Agent compaction policy requires at least one strategy"
        )
    if self.trigger_token_threshold <= 0:
        raise AgentDefinitionError(
            "Agent compaction trigger token threshold must be positive"
        )

AgentExecutionSpec(name=None, objective=None, instructions=None, output_type=None, accepted_signals=(), recovery=RecoveryStrategy.NONE, streaming_exposure_mode=StreamingExposureMode.BALANCED, timeout_seconds=None, limits=AgentExecutionLimits(), teammates=(), compaction=None, delegation_allowed=False, metadata=dict()) dataclass

Declarative execution semantics that cannot be inferred from DI alone.

__post_init__()

Reject execution specs that would fail later at bootstrap.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def __post_init__(self) -> None:
    """Reject execution specs that would fail later at bootstrap."""
    if self.name is not None and not self.name.strip():
        raise AgentDefinitionError("Agent name cannot be blank")
    if self.objective is not None and not self.objective.strip():
        raise AgentDefinitionError("Agent objective cannot be blank")
    if self.instructions is not None and not self.instructions.strip():
        raise AgentDefinitionError("Agent instructions cannot be blank")
    if self.output_type is not None and not isclass(self.output_type):
        raise AgentDefinitionError("Agent output type must be a class")
    if self.timeout_seconds is not None and self.timeout_seconds <= 0:
        raise AgentDefinitionError("Agent timeout must be positive")
    if (
        self.timeout_seconds is not None
        and self.limits.timeout_seconds is not None
        and self.timeout_seconds != self.limits.timeout_seconds
    ):
        raise AgentDefinitionError("Agent timeout declarations must match")
    teammate_names = [teammate.name for teammate in self.teammates]
    if len(set(teammate_names)) != len(teammate_names):
        raise AgentDefinitionError("Agent teammate names must be unique")

Agent(spec=AgentExecutionSpec(), *, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

UseCase-equivalent Pod stereotype for agentic workflow components.

validate_bootstrap()

Re-run definition validation during application bootstrap.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def validate_bootstrap(self) -> None:
    """Re-run definition validation during application bootstrap."""
    self._validate_execute_contract(self._ensure_agent_class(self.target))

required_persistence_repository_types()

Return repository ports required by this Agent's durable path.

Source code in core/spakky-agent/src/spakky/agent/execution.py

def required_persistence_repository_types(self) -> tuple[type[object], ...]:
    """Return repository ports required by this Agent's durable path."""
    from spakky.agent.interfaces.repository import (
        IAgentEvidenceRepository,
        IAgentSignalRepository,
        IAgentStateRepository,
    )

    if (
        self.spec.recovery is RecoveryStrategy.ACTION_BOUNDARY
        or len(self.spec.accepted_signals) > 0
    ):
        return (
            IAgentStateRepository,
            IAgentSignalRepository,
            IAgentEvidenceRepository,
        )
    return ()

Inbound run contract consumed by the framework-owned agent runner.

ADR-0013 §1 hands the execution loop to the framework runner. A caller (an inbound adapter today, an AG-UI/A2A protocol adapter later) hands the runner a RunAgentInput describing one run: which durable run to correlate against, the user instruction that seeds the model request, and whether the run resumes a paused/interrupted execution (ADR-0013 §5 HITL resume / restart recovery).

A stateless caller may also carry the prior transcript inline through message_history (ADR-0013 §6 client-injected history, pydantic-ai message_history precedent). When it is empty and a TaskStore is wired, the runner instead loads the server-persisted transcript by effective_conversation_id — the two multi-turn paths are mutually exclusive per run, never merged.

Approval decisions are not carried here. The unified pause -> approval request -> resume flow (ADR-0013 §5) delivers a decision through the durable signal repository, not through this input, so the runner polls the signal queue non-blockingly rather than reading a decision off the inbound contract.

RunAgentInput(state_id, instruction, conversation_id=None, parent_run_id=None, resume=False, message_history=(), model_selection=None, metadata=dict()) dataclass

Inbound contract for one framework-owned agent run.

effective_conversation_id property

Return the multi-turn thread id, defaulting to the run id.

ADR-0013 §3 attribution requires a conversation id on every event. A single-turn caller may omit it, in which case the run id identifies the (degenerate) one-turn conversation.

__post_init__()

Reject inbound input that cannot correlate a run or seed a request.

Source code in core/spakky-agent/src/spakky/agent/inbound.py

def __post_init__(self) -> None:
    """Reject inbound input that cannot correlate a run or seed a request."""
    if not self.state_id.strip():
        raise AgentDefinitionError("Run agent input state id cannot be blank")
    if not self.instruction.strip():
        raise AgentDefinitionError("Run agent input instruction cannot be blank")
    if self.conversation_id is not None and not self.conversation_id.strip():
        raise AgentDefinitionError(
            "Run agent input conversation id cannot be blank"
        )
    if self.parent_run_id is not None and not self.parent_run_id.strip():
        raise AgentDefinitionError("Run agent input parent run id cannot be blank")

Framework-owned agent execution loop (ADR-0013 §1).

The runner generalizes the manual model -> tool -> evidence -> terminate loop that ADR-0009 left in developer execute() bodies. A developer declares an @Agent spec plus @agent_tool methods; the framework provides the loop. The runner consumes the provider-neutral model stream (C2), dispatches tool calls through the discovered catalog (C4), records boundary/evidence (C3 model contracts), consumes durable signals, drives the unified HITL pause -> approval request -> resume flow (ADR-0013 §5), and terminates with a typed final output shaped by spec.output_type.

The runner exposes two streams over the same orchestration. run() yields the public AgentYield vocabulary that inbound adapters already consume. run_events() yields the protocol-neutral AgentEvent taxonomy (ADR-0013 §3) that AG-UI (#414) and A2A (#415) adapters project losslessly: the runner emits message/reasoning deltas, the tool-call start/args-delta/end/result lifecycle, and run/step boundaries as distinct events carrying attribution (agent / run / parent / conversation), rather than collapsing them into coarse AgentYield items that an adapter would have to re-expand. The fine-grained model-stream channels (C2 ModelStreamEventKind message/reasoning/tool-args deltas) project one-to-one onto the neutral taxonomy; REASONING_DELTA is omitted when the model declares no reasoning capability (graceful degrade, ADR-0013 §4).

DEFAULT_SYSTEM_INSTRUCTION = 'Use the declared tools to accomplish the objective.' module-attribute

Fallback system message when an agent spec declares no instructions.

DEFAULT_SAMPLING = SamplingOptions(temperature=0.0, max_tokens=512) module-attribute

Deterministic default sampling for the framework-owned model request.

ESTIMATED_CHARACTERS_PER_TOKEN = 4 module-attribute

Provider-neutral characters-per-token ratio for the compaction trigger estimate.

The core cannot call a provider tokenizer (it stays protocol-neutral, ADR-0013 §2), so the compaction trigger estimates token count from transcript length using the widely-used ~4-characters-per-token approximation. The estimate only decides whether the declared chain runs; the strategies themselves bound the result, so an approximate trigger is sufficient.

AgentRunResult(state_id, status, tool_calls, evidence_count) dataclass

Neutral terminal summary returned when a spec declares no output type.

AgentRunner(agent, target, model, states=None, signals=None, evidence=None, task_store=None) dataclass

Framework-owned standard agent loop bound to one agent instance.

Durable repositories are None for a stateless agent (one that declares no accepted_signals and no action-boundary recovery). In that mode the runner skips state/evidence/signal handling and runs model -> tool -> final.

for_agent_instance(instance) classmethod

Resolve runner ports from an agent instance's injected attributes.

ADR-0009 stores constructor-injected ports as instance attributes, so the runner reads vars(instance) (the typed instance __dict__, not the banned getattr) and resolves each port by runtime type. Attribute names are developer-chosen, so resolution is type-driven, not name-driven.

Source code in core/spakky-agent/src/spakky/agent/runner.py

@classmethod
def for_agent_instance(cls, instance: object) -> "AgentRunner":
    """Resolve runner ports from an agent instance's injected attributes.

    ADR-0009 stores constructor-injected ports as instance attributes, so the
    runner reads ``vars(instance)`` (the typed instance ``__dict__``, not the
    banned ``getattr``) and resolves each port by runtime type. Attribute
    names are developer-chosen, so resolution is type-driven, not name-driven.
    """
    agent = Agent.get(type(instance))
    attributes = tuple(vars(instance).values())
    model = cls._resolve_required(attributes, IAgentModel)
    if model is None:
        raise AgentModelConfigurationError(
            "Agent run requires an IAgentModel port but none was injected"
        )
    states = cls._resolve_optional(attributes, IAgentStateRepository)
    signals = cls._resolve_optional(attributes, IAgentSignalRepository)
    evidence = cls._resolve_optional(attributes, IAgentEvidenceRepository)
    task_store = cls._resolve_optional(attributes, ITaskStore)
    runner = cls(
        agent=agent,
        target=instance,
        model=model,
        states=states,
        signals=signals,
        evidence=evidence,
        task_store=task_store,
    )
    runner._require_durable_ports()
    return runner

with_model(model)

Return a runner using a run-specific model adapter.

Source code in core/spakky-agent/src/spakky/agent/runner.py

def with_model(self, model: IAgentModel) -> "AgentRunner":
    """Return a runner using a run-specific model adapter."""
    return replace(self, model=model)

run(run_input) async

Run one model-mediated agent loop, yielding the public stream.

Source code in core/spakky-agent/src/spakky/agent/runner.py

async def run(
    self,
    run_input: RunAgentInput,
) -> AsyncGenerator[AgentYield[object], None]:
    """Run one model-mediated agent loop, yielding the public stream."""
    if self.states is None:
        async for item in self._run_stateless(run_input):
            yield item
        return
    async for item in self._run_durable(run_input, self.states):
        yield item

run_events(run_input) async

Run one model-mediated loop, emitting the neutral event taxonomy.

This is the lossless native stream AG-UI/A2A adapters consume (ADR-0013 §3). It shares the run's orchestration with run() — model request, tool dispatch, durable approval gating, evidence persistence — but emits distinct AgentEvent members instead of coarse AgentYield items, so an adapter projects each event one-to-one rather than re-expanding framing.

Source code in core/spakky-agent/src/spakky/agent/runner.py

async def run_events(
    self,
    run_input: RunAgentInput,
) -> AsyncGenerator[AgentEvent, None]:
    """Run one model-mediated loop, emitting the neutral event taxonomy.

    This is the lossless native stream AG-UI/A2A adapters consume (ADR-0013
    §3). It shares the run's orchestration with ``run()`` — model request,
    tool dispatch, durable approval gating, evidence persistence — but emits
    distinct ``AgentEvent`` members instead of coarse ``AgentYield`` items, so
    an adapter projects each event one-to-one rather than re-expanding framing.
    """
    attribution = self._event_attribution(run_input)
    state = (
        self._ensure_active_state(run_input, self.states)
        if self.states is not None
        else None
    )
    cursor = _MessageCursor(state_id=run_input.state_id)
    history = await self._resolved_history(run_input)
    assistant_text: list[str] = []
    yield RunStartedEvent(attribution)
    if state is not None:
        cancel = await self._consume_cancel(state)
        if cancel is not None:
            yield RunFinishedEvent(attribution, error=_cancel_error(cancel))
            return
    yield StepStartedEvent(attribution, step_name="model-call")
    async for event in self.model.stream(self._model_request(run_input, history)):
        if state is not None:
            cancel = await self._consume_cancel(state)
            if cancel is not None:
                yield StepFinishedEvent(attribution, step_name="model-call")
                yield RunFinishedEvent(attribution, error=_cancel_error(cancel))
                return
        async for item in self._consume_stream_event_as_events(
            event, state, attribution, cursor, run_input
        ):
            yield item
        self._accumulate_assistant_text(event, assistant_text)
        if event.kind is ModelStreamEventKind.ERROR and event.error is not None:
            yield StepFinishedEvent(attribution, step_name="model-call")
            yield RunFinishedEvent(
                attribution,
                error={"code": event.error.code, "message": event.error.message},
            )
            return
        if (
            state is not None
            and event.kind is ModelStreamEventKind.TOOL_CALL_CANDIDATE
            and event.tool_call is not None
            and self._states_required().get(state.id).status
            is not AgentStatus.ACTIVE
        ):
            current = self._states_required().get(state.id)
            yield StepFinishedEvent(attribution, step_name="model-call")
            if current.status is AgentStatus.INTERRUPTED:
                yield _run_paused_event(current, attribution)
                return
            yield RunFinishedEvent(attribution, error=_state_error(current))
            return
    yield StepFinishedEvent(attribution, step_name="model-call")
    # A tool that paused for approval already saved a WAIT_FOR_APPROVAL state
    # (INTERRUPTED, not ACTIVE); completing it would clobber the durable pause
    # the adapter surfaces as its deferred-tool request, so only an unpaused
    # (still ACTIVE) run transitions to COMPLETED — mirroring _run_durable.
    if (
        state is not None
        and self._states_required().get(state.id).status is AgentStatus.ACTIVE
    ):
        self._complete_state(state.id)
    elif state is not None:
        current = self._states_required().get(state.id)
        if current.status is AgentStatus.INTERRUPTED:
            yield _run_paused_event(current, attribution)
            return
        yield RunFinishedEvent(attribution, error=_state_error(current))
        return
    self._persist_turns(run_input, assistant_text)
    yield RunFinishedEvent(attribution)

runner_backed_execute(self, run_input) async

Framework-provided execute() bound onto a declaration-only @Agent.

@Agent binds this as the agent's execute() when the developer declares only a spec plus @agent_tool methods (ADR-0013 §1). It builds a runner from the agent instance's injected ports and replays its stream.

Source code in core/spakky-agent/src/spakky/agent/runner.py

async def runner_backed_execute(
    self: object,
    run_input: RunAgentInput,
) -> AsyncGenerator[AgentYield[object], None]:
    """Framework-provided ``execute()`` bound onto a declaration-only @Agent.

    ``@Agent`` binds this as the agent's ``execute()`` when the developer
    declares only a spec plus ``@agent_tool`` methods (ADR-0013 §1). It builds a
    runner from the agent instance's injected ports and replays its stream.
    """
    runner = AgentRunner.for_agent_instance(self)
    async for item in runner.run(run_input):
        yield item

DI entry point for opening agent runners.

Protocol adapters should not construct :class:AgentRunner directly when an application may load runner-augmenting plugins such as MCP. They ask this factory port for a runner scoped to one request instead; plugins can then bind the port to their own implementation while the default remains the native framework runner.

IAgentRunnerFactory

Bases: ABC

Port for opening a request-scoped runner for one agent instance.

open_runner(agent_instance, run_input=None) abstractmethod

Yield a runner bound to agent_instance for one adapter request.

Source code in core/spakky-agent/src/spakky/agent/runner_factory.py

@abstractmethod
def open_runner(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> AbstractAsyncContextManager[AgentRunner]:
    """Yield a runner bound to ``agent_instance`` for one adapter request."""
    ...

AgentRunnerFactory(model_resolver=None)

Bases: IAgentRunnerFactory

Default runner factory using the native framework-owned runner.

Source code in core/spakky-agent/src/spakky/agent/runner_factory.py

def __init__(self, model_resolver: IAgentModelResolver | None = None) -> None:
    self._model_resolver = model_resolver

open_runner(agent_instance, run_input=None) async

Yield the native runner for one request.

Source code in core/spakky-agent/src/spakky/agent/runner_factory.py

@override
@asynccontextmanager
async def open_runner(
    self,
    agent_instance: object,
    run_input: RunAgentInput | None = None,
) -> AsyncGenerator[AgentRunner, None]:
    """Yield the native runner for one request."""
    runner = AgentRunner.for_agent_instance(agent_instance)
    if self._model_resolver is not None:
        model = self._model_resolver.resolve_model(agent_instance, run_input)
        if model is not None:
            runner = runner.with_model(model)
    yield runner

Event

Protocol-neutral agent event taxonomy.

This taxonomy is the single neutral source that AG-UI, A2A, and MCP adapters normalize into their respective protocols. The core stays protocol-neutral (ADR-0013 §2 / §3): no event field is named after, or shaped by, any single external protocol, yet every event carries enough attribution for each adapter to reconstruct that protocol's linkage without loss.

Attribution carried by every event (ADR-0013 §3):

• agent_id — which agent emitted the event (team-mode composite attribution).
• run_id — the run this event belongs to. AG-UI projects this as runId; A2A projects this as the A2A Task id.
• parent_run_id — the parent run in a delegation tree (None for a root run that has no delegating parent). AG-UI projects this as parentRunId; A2A projects this as the parent task in its task hierarchy.
• conversation_id — the multi-turn conversation this run participates in. AG-UI projects this as threadId; A2A projects this as contextId.

AgentEvent = MessageDeltaEvent | ReasoningDeltaEvent | ToolCallStartEvent | ToolCallArgsDeltaEvent | ToolCallEndEvent | ToolCallResultEvent | RunStartedEvent | RunPausedEvent | RunFinishedEvent | StepStartedEvent | StepFinishedEvent | StateSnapshotEvent | StateDeltaEvent | ArtifactEvent

Discriminated union of every protocol-neutral agent event.

Each member exposes a distinct kind literal so adapters dispatch with match/case exhaustiveness rather than runtime attribute probing.

AgentEventKind

Bases: StrEnum

Protocol-neutral event kinds emitted across an agent run.

The lifecycle kinds (RUN_STARTED … ARTIFACT) generalize ADR-0009's public AgentYield vocabulary and align 1:1 with the streaming events that AG-UI and A2A adapters must reproduce (ADR-0013 §3).

AgentEventAttribution(agent_id, run_id, conversation_id, parent_run_id=None) dataclass

Attribution carried by every protocol-neutral agent event.

parent_run_id is None exactly when the run is a delegation-tree root (ADR-0013 §3 "parent link" / ADR-0009 parent linkage). The remaining ids are always present so adapters never have to synthesize linkage.

__post_init__()

Reject attribution that cannot identify an agent, run, or conversation.

Source code in core/spakky-agent/src/spakky/agent/event.py

def __post_init__(self) -> None:
    """Reject attribution that cannot identify an agent, run, or conversation."""
    if not self.agent_id.strip():
        raise AgentDefinitionError("Agent event agent id cannot be blank")
    if not self.run_id.strip():
        raise AgentDefinitionError("Agent event run id cannot be blank")
    if not self.conversation_id.strip():
        raise AgentDefinitionError("Agent event conversation id cannot be blank")
    if self.parent_run_id is not None and not self.parent_run_id.strip():
        raise AgentDefinitionError("Agent event parent run id cannot be blank")

MessageDeltaEvent(attribution, message_id, delta, metadata=dict()) dataclass

Incremental assistant message text produced by the model.

ReasoningDeltaEvent(attribution, reasoning_id, delta, metadata=dict()) dataclass

Incremental model reasoning (thinking) text.

ToolCallStartEvent(attribution, call_id, tool_name, parent_message_id=None, metadata=dict()) dataclass

A model has begun emitting a tool call with a known name.

parent_message_id links the tool call to the assistant message that requested it (None when the model emits the call outside any message). AG-UI projects this as parentMessageId on TOOL_CALL_START.

ToolCallArgsDeltaEvent(attribution, call_id, args_delta, metadata=dict()) dataclass

Incremental serialized arguments for an in-flight tool call.

ToolCallEndEvent(attribution, call_id, metadata=dict()) dataclass

A tool call has finished streaming its arguments.

ToolCallResultEvent(attribution, call_id, tool_name, message_id, result=None, metadata=dict()) dataclass

The result returned by executing a completed tool call.

message_id identifies the conversation message this tool result belongs to, so an adapter can attach the result to the right message. AG-UI requires it as messageId on TOOL_CALL_RESULT; without it the result frame cannot be reconstructed losslessly.

RunStartedEvent(attribution, metadata=dict()) dataclass

A run has started executing.

RunPausedEvent(attribution, reason, prompt, state_id, approval_id=None, tool_call_id=None, allowed_decisions=(), metadata=dict()) dataclass

A run has paused without terminal success or failure.

The event is intentionally protocol-neutral: it carries core lifecycle reason plus the input prompt/decision envelope adapters need to project their own input-required or auth-required protocol states.

__post_init__()

Reject pause events that cannot be shown or resumed.

Source code in core/spakky-agent/src/spakky/agent/event.py

def __post_init__(self) -> None:
    """Reject pause events that cannot be shown or resumed."""
    if not self.prompt.strip():
        raise AgentDefinitionError("Agent run pause prompt cannot be blank")
    if not self.state_id.strip():
        raise AgentDefinitionError("Agent run pause state id cannot be blank")
    if self.approval_id is not None and not self.approval_id.strip():
        raise AgentDefinitionError("Agent run pause approval id cannot be blank")
    if self.tool_call_id is not None and not self.tool_call_id.strip():
        raise AgentDefinitionError("Agent run pause tool call id cannot be blank")

RunFinishedEvent(attribution, error=None, metadata=dict()) dataclass

A run has finished executing.

error is None for a successful run and carries a terminal failure payload otherwise (A2A failed task / AG-UI RUN_ERROR).

StepStartedEvent(attribution, step_name, metadata=dict()) dataclass

A named step inside a run has started (one model-loop iteration).

StepFinishedEvent(attribution, step_name, metadata=dict()) dataclass

A named step inside a run has finished.

StateSnapshotEvent(attribution, snapshot, metadata=dict()) dataclass

A full snapshot of shared run state.

StateDeltaEvent(attribution, patch, metadata=dict()) dataclass

An incremental change to shared run state.

patch is a sequence of JSON-Patch operations (RFC 6902), the same shape AG-UI STATE_DELTA carries, so adapters relay it without re-deriving a diff.

ArtifactEvent(attribution, artifact_id, content, name=None, metadata=dict()) dataclass

A produced artifact surfaced by an agent run.

Dispatcher

Declarative tool dispatch over a discovered agent tool catalog.

ADR-0013 §1 hands the model-call -> tool-invoke step to the framework runner. This module removes the developer-written if call.name == ... chain and the manual payload extraction it implied: a model tool call is resolved against the catalog, its arguments are bound through the descriptor, and the descriptor's callable is invoked. External MCP tools (follow-up F1) normalize into the same AgentToolCatalog and therefore dispatch through this identical path.

AgentToolDispatcher(target, catalog, runtime_context=None) dataclass

Resolve and invoke a catalog tool from a model tool call.

The dispatcher binds to a single agent instance whose @agent_tool methods are described by catalog. Catalog descriptors that own no instance parameter (such as MCP-normalized external tools) are invoked without the bound target.

descriptor_for(call)

Resolve the catalog descriptor a model tool call targets.

Source code in core/spakky-agent/src/spakky/agent/dispatcher.py

def descriptor_for(self, call: ModelToolCall) -> AgentToolDescriptor:
    """Resolve the catalog descriptor a model tool call targets."""
    for descriptor in self.catalog.descriptors:
        if descriptor.schema.name == call.name:
            return descriptor
    raise AgentToolDispatchError("Agent tool call names an unregistered tool")

dispatch(call) async

Bind a model tool call payload and invoke its catalog callable.

Source code in core/spakky-agent/src/spakky/agent/dispatcher.py

async def dispatch(self, call: ModelToolCall) -> object:
    """Bind a model tool call payload and invoke its catalog callable."""
    descriptor = self.descriptor_for(call)
    invocation = descriptor.bind_invocation(call.arguments)
    callable_ = descriptor.callable
    positional = self._with_injected_prefix(callable_, invocation.args)
    if iscoroutinefunction(callable_):
        awaitable: Awaitable[object] = callable_(
            *positional,
            **invocation.kwargs,
        )
        return await awaitable
    return callable_(*positional, **invocation.kwargs)

State

Agent lifecycle state contracts.

AgentStatus

Bases: StrEnum

Externally observable lifecycle states for an agent execution.

AgentStateTransition

Bases: StrEnum

State transition vocabulary accepted by durable agent orchestration.

AgentStateReason

Bases: StrEnum

Structured reason that refines an externally observable lifecycle state.

AgentState(id, agent_type, status, transition=None, reason=None, current_activity=None, input_ref=None, output_ref=None, pending_signal_count=0, last_event_cursor=None, recovery_marker=None, metadata=dict(), created_at=None, updated_at=None) dataclass

Materialized state for a long-running agent execution.

__post_init__()

Reject state snapshots that cannot represent a real queue count.

Source code in core/spakky-agent/src/spakky/agent/state.py

def __post_init__(self) -> None:
    """Reject state snapshots that cannot represent a real queue count."""
    if self.pending_signal_count < 0:
        raise AgentDefinitionError("Agent pending signal count cannot be negative")

Signal

Agent signal contracts.

AgentSignalKind

Bases: StrEnum

Inbound stimulus kinds that an agent may accept while running.

ApprovalDecision

Bases: StrEnum

Human-in-the-loop approval outcomes.

AgentSignal(id, agent_state_id, kind, payload=dict(), created_at=None) dataclass

Inbound stimulus appended for an agent execution.

Non-blocking consumption helpers for durable agent signals.

AgentSignalPollPoint

Bases: StrEnum

Places where orchestration may poll durable signals without waiting.

AgentSignalConsumptionBatch(state_id, poll_point, signals) dataclass

Signals consumed during one non-blocking poll.

consumed_count property

Return the number of signals consumed by this poll.

consume_pending_agent_signals(repository, state_id, *, poll_point=AgentSignalPollPoint.SAFE_BOUNDARY, accepted_signals=None, max_signals=None)

Consume currently pending signals without sleeping or waiting for new input.

The repository remains responsible for durable queue ordering. This helper consumes only the eligible prefix of the pending queue so later signals never overtake an earlier unaccepted signal.

Source code in core/spakky-agent/src/spakky/agent/signal_consumption.py

def consume_pending_agent_signals(
    repository: IAgentSignalRepository,
    state_id: str,
    *,
    poll_point: AgentSignalPollPoint = AgentSignalPollPoint.SAFE_BOUNDARY,
    accepted_signals: Sequence[AgentSignalKind] | None = None,
    max_signals: int | None = None,
) -> AgentSignalConsumptionBatch:
    """Consume currently pending signals without sleeping or waiting for new input.

    The repository remains responsible for durable queue ordering. This helper
    consumes only the eligible prefix of the pending queue so later signals never
    overtake an earlier unaccepted signal.
    """
    if not state_id.strip():
        raise AgentDefinitionError("Agent signal state id cannot be blank")
    if max_signals is not None and max_signals <= 0:
        raise AgentDefinitionError("Agent signal max_signals must be positive")

    accepted = frozenset(accepted_signals) if accepted_signals is not None else None
    pending = tuple(repository.list_pending(state_id))
    selected = _select_consumable_prefix(
        pending,
        accepted_signals=accepted,
        max_signals=max_signals,
    )
    consumed = tuple(repository.mark_consumed(signal.id) for signal in selected)

    return AgentSignalConsumptionBatch(
        state_id=state_id,
        poll_point=poll_point,
        signals=consumed,
    )

Evidence

Agent evidence contracts.

AgentEvidenceKind

Bases: StrEnum

Kinds of append-only evidence captured by agent execution.

AgentEvidence(id, agent_state_id, kind, payload=dict(), summary=None, digest=None, manifest_ref=None, reference=None, sensitive_fields=(), created_at=None) dataclass

Append-only artifact captured during an agent execution.

AgentEvidenceCandidate(kind, payload=dict(), summary=None, digest=None, manifest_ref=None, reference=None, sensitive_fields=()) dataclass

Append-only evidence candidate before repository id assignment.

tool_result(*, tool_identity, tool_schema_name, result, capture, sensitive_fields=(), exposure_policy=None, summary=None) classmethod

Create evidence metadata for a captured tool result.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def tool_result(
    cls,
    *,
    tool_identity: str,
    tool_schema_name: str,
    result: JsonObject,
    capture: EvidenceCapture,
    sensitive_fields: tuple[SensitiveFieldDescriptor, ...] = (),
    exposure_policy: EvidenceExposurePolicy | None = None,
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create evidence metadata for a captured tool result."""
    _require_non_blank(tool_identity, "Agent tool identity")
    _require_non_blank(tool_schema_name, "Agent tool schema name")
    policy = exposure_policy or EvidenceExposurePolicy()
    guarded_result = guard_json_value(result, sensitive_fields, policy)
    if not isinstance(guarded_result, Mapping):
        guarded_result = {}
    return cls(
        kind=AgentEvidenceKind.TOOL,
        payload={
            "tool_identity": tool_identity,
            "tool_schema_name": tool_schema_name,
            "capture": capture.value,
            "result": guarded_result,
        },
        sensitive_fields=sensitive_fields
        if policy.include_sensitive_metadata
        else (),
        summary=summary,
    )

model_decision(*, model, decision, summary=None) classmethod

Create evidence metadata for a model decision.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def model_decision(
    cls,
    *,
    model: str,
    decision: JsonObject,
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create evidence metadata for a model decision."""
    _require_non_blank(model, "Agent model name")
    return cls(
        kind=AgentEvidenceKind.MODEL,
        payload={"model": model, "decision": decision},
        summary=summary,
    )

tool_decision(*, tool_identity, decision, summary=None) classmethod

Create evidence metadata for a tool-routing decision.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def tool_decision(
    cls,
    *,
    tool_identity: str,
    decision: JsonObject,
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create evidence metadata for a tool-routing decision."""
    _require_non_blank(tool_identity, "Agent tool identity")
    return cls(
        kind=AgentEvidenceKind.TOOL,
        payload={"tool_identity": tool_identity, "decision": decision},
        summary=summary,
    )

context_optimization(*, action, stage, signals=(), summary=None) classmethod

Create before/after evidence for a context optimization action.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

@classmethod
def context_optimization(
    cls,
    *,
    action: ContextOptimizationAction,
    stage: ContextOptimizationEvidenceStage,
    signals: Sequence[ContextHealthSignal] = (),
    summary: str | None = None,
) -> "AgentEvidenceCandidate":
    """Create before/after evidence for a context optimization action."""
    _require_non_blank(action.id, "Context optimization action id")
    return cls(
        kind=AgentEvidenceKind.CONTEXT_OPTIMIZATION,
        payload={
            "stage": stage.value,
            "action": action.evidence_payload(),
            "signals": tuple(signal.evidence_payload() for signal in signals),
        },
        summary=summary,
        digest=action.digest_ref,
        manifest_ref=action.manifest_ref,
        reference=action.result_evidence_ref,
    )

to_evidence(*, evidence_id, agent_state_id, created_at=None)

Assign repository identity while preserving append-only contents.

Source code in core/spakky-agent/src/spakky/agent/evidence.py

def to_evidence(
    self,
    *,
    evidence_id: str,
    agent_state_id: str,
    created_at: datetime | None = None,
) -> AgentEvidence:
    """Assign repository identity while preserving append-only contents."""
    _require_non_blank(evidence_id, "Agent evidence id")
    _require_non_blank(agent_state_id, "Agent state id")
    return AgentEvidence(
        id=evidence_id,
        agent_state_id=agent_state_id,
        kind=self.kind,
        payload=self.payload,
        summary=self.summary,
        digest=self.digest,
        manifest_ref=self.manifest_ref,
        reference=self.reference,
        sensitive_fields=self.sensitive_fields,
        created_at=created_at,
    )

Context

Typed context contracts for agent model input assembly.

ContextPackRole

Bases: StrEnum

Semantic role of a context pack inside a model request.

ContextFreshness

Bases: StrEnum

Freshness classification for context rot and budget decisions.

ContextSensitivity

Bases: StrEnum

Deterministic sensitivity metadata carried before model input.

ContextRotSymptom

Bases: StrEnum

Typed context rot symptoms observed before model input assembly.

ContextOptimizationActionKind

Bases: StrEnum

Optimization actions that can be selected from context health signals.

ContextOptimizationEvidenceStage

Bases: StrEnum

Where an optimization action evidence item sits in the agent flow.

ContextTokenBudget(max_tokens=None, estimated_tokens=None, reserved_output_tokens=None) dataclass

Token budget allocated to one context pack.

ContextHealthSignal(id, symptom, manifest_ref=None, pack_id=None, evidence_ref=None, score=None, observed_at=None, metadata=dict()) dataclass

Observed context rot signal used to choose optimization actions.

evidence_payload()

Return JSON-compatible signal metadata for append-only evidence.

Source code in core/spakky-agent/src/spakky/agent/context.py

def evidence_payload(self) -> Mapping[str, JsonValue]:
    """Return JSON-compatible signal metadata for append-only evidence."""
    return {
        "id": self.id,
        "symptom": self.symptom.value,
        "manifest_ref": self.manifest_ref,
        "pack_id": self.pack_id,
        "evidence_ref": self.evidence_ref,
        "score": self.score,
        "observed_at": self.observed_at.isoformat()
        if self.observed_at is not None
        else None,
        "metadata": self.metadata,
    }

ContextOptimizationAction(id, kind, signal_refs=tuple(), target_pack_ids=tuple(), manifest_ref=None, digest_ref=None, delegation_ref=None, result_evidence_ref=None, reason=None, metadata=dict()) dataclass

Selected optimization action derived from context health signals.

evidence_payload()

Return JSON-compatible action metadata without raw context contents.

Source code in core/spakky-agent/src/spakky/agent/context.py

def evidence_payload(self) -> Mapping[str, JsonValue]:
    """Return JSON-compatible action metadata without raw context contents."""
    return {
        "id": self.id,
        "kind": self.kind.value,
        "signal_refs": tuple(self.signal_refs),
        "target_pack_ids": tuple(self.target_pack_ids),
        "manifest_ref": self.manifest_ref,
        "digest_ref": self.digest_ref,
        "delegation_ref": self.delegation_ref,
        "result_evidence_ref": self.result_evidence_ref,
        "reason": self.reason,
        "metadata": self.metadata,
    }

IAgentContextHandler

Bases: ABC

Select context optimization actions from health signals and manifests.

select_optimization_actions(signals, manifest) abstractmethod

Return optimization actions without mutating raw evidence.

Source code in core/spakky-agent/src/spakky/agent/context.py

@abstractmethod
def select_optimization_actions(
    self,
    signals: Sequence[ContextHealthSignal],
    manifest: "ContextManifest",
) -> Sequence[ContextOptimizationAction]:
    """Return optimization actions without mutating raw evidence."""
    ...

ContextPack(id, content, source, role, freshness=ContextFreshness.UNKNOWN, relevance=None, token_budget=ContextTokenBudget(), sensitivity=ContextSensitivity.INTERNAL, sensitive_fields=(), metadata=dict()) dataclass

LLM-facing context unit derived from state, signal, or evidence.

guarded_content(policy=None)

Return deterministic model-safe content for this context pack.

Source code in core/spakky-agent/src/spakky/agent/context.py

def guarded_content(
    self,
    policy: ContextExposurePolicy | None = None,
) -> str:
    """Return deterministic model-safe content for this context pack."""
    exposure_policy = policy or ContextExposurePolicy()
    if self.sensitivity == ContextSensitivity.REDACTED:
        return "[REDACTED]"
    guarded = guard_json_value(
        {"content": self.content},
        tuple(
            SensitiveFieldDescriptor(
                ("content", *descriptor.path), descriptor.field
            )
            for descriptor in self.sensitive_fields
        ),
        exposure_policy,
    )
    content = cast(Mapping[str, JsonValue], guarded).get("content")
    if isinstance(content, str):
        return content
    return "[REDACTED]"

message_metadata(policy=None)

Return non-content metadata for provider-neutral model messages.

Source code in core/spakky-agent/src/spakky/agent/context.py

def message_metadata(
    self,
    policy: ContextExposurePolicy | None = None,
) -> Mapping[str, JsonValue]:
    """Return non-content metadata for provider-neutral model messages."""
    exposure_policy = policy or ContextExposurePolicy()
    metadata: dict[str, JsonValue] = {
        "context_pack_id": self.id,
        "source": self.source,
        "role": self.role.value,
        "freshness": self.freshness.value,
        "relevance": self.relevance,
        "token_budget": {
            "max_tokens": self.token_budget.max_tokens,
            "estimated_tokens": self.token_budget.estimated_tokens,
            "reserved_output_tokens": self.token_budget.reserved_output_tokens,
        },
        "sensitivity": self.sensitivity.value,
        "metadata": self.metadata,
    }
    if exposure_policy.include_sensitive_context_metadata and self.sensitive_fields:
        metadata["sensitive_fields"] = tuple(
            descriptor.to_metadata() for descriptor in self.sensitive_fields
        )
    return metadata

ContextManifestEntry(pack_id, source, role, origin_ref, evidence_ref=None, digest_ref=None, sensitive_fields=(), metadata=dict()) dataclass

One audited pack entry inside a context manifest.

ContextManifest(id, entries, origin_ref=None, evidence_refs=tuple(), created_at=None, metadata=dict()) dataclass

Auditable composition record for model input context packs.

ContextDigest(id, context_identity, source_manifest_ref, digest, derived_from_pack_ids=tuple(), compression_evidence_ref=None, algorithm=None, summary=None, created_at=None, metadata=dict()) dataclass

Derived compression evidence for a context identity.

Compaction

Pluggable context-compaction strategies (ADR-0013 §7).

A long conversation eventually outgrows a model backend's context window. The framework runner owns when to compact (ADR-0013 §1 declarative loop ownership); this module owns how. ICompactionStrategy is the provider-neutral port a developer declares in an @Agent spec, and the runner applies the declared chain to the resolved history before each model request once the running token estimate crosses the policy threshold.

The port shape follows pydantic-ai's message-history processor / ProcessHistory capability: a processor receives the message list (plus run context such as token usage) and returns a transformed list. Here the contextual inputs are made explicit parameters — the running ModelUsage and the backend ModelCapability — so a strategy decides how aggressively to compact from the same signals the runner uses to decide whether to compact. Strategies are pure transforms over tuple[ModelMessage, ...] and compose by sequential application, so a chain of strategies is just each applied to the previous one's output.

SUMMARY_MESSAGE_METADATA_KEY = 'compaction' module-attribute

Metadata flag marking a message the summarize strategy synthesized.

SUMMARY_MESSAGE_METADATA_VALUE = 'summary' module-attribute

Metadata value identifying a synthesized transcript summary message.

DEFAULT_SUMMARY_INSTRUCTION = 'Summarize the earlier conversation turns below into a concise briefing that preserves decisions, facts, and open questions. Reply with the summary only.' module-attribute

Fallback instruction for the secondary model that summarizes old turns.

ICompactionStrategy

Bases: ABC

Provider-neutral context-compaction transform applied before a request.

A strategy maps a resolved history to a shorter one, reading the running ModelUsage and backend ModelCapability so it can scale its effect to how close the run is to the context limit. Implementations are pure transforms — they never mutate the input — so the runner can apply a declared chain by threading each strategy's output into the next.

compact(history, usage, capability) abstractmethod async

Return a compacted view of history for the next model request.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

@abstractmethod
async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Return a compacted view of ``history`` for the next model request."""
    ...

KeepRecentMessagesCompactionStrategy(max_messages) dataclass

Bases: ICompactionStrategy

Sliding-window strategy that keeps only the most recent messages.

The cheapest compaction: drop the oldest messages and keep the last max_messages. It carries no model dependency, so it is the safe default tail of a chain that bounds history length regardless of content.

__post_init__()

Reject a window that would keep no messages.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

def __post_init__(self) -> None:
    """Reject a window that would keep no messages."""
    if self.max_messages <= 0:
        raise AgentDefinitionError(
            "Keep-recent compaction window must keep at least one message"
        )

compact(history, usage, capability) async

Keep only the last max_messages messages of the history.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Keep only the last ``max_messages`` messages of the history."""
    return history[-self.max_messages :]

TrimToolResultsCompactionStrategy(max_characters) dataclass

Bases: ICompactionStrategy

Truncate verbose tool-result messages while preserving the dialogue.

Tool results (search dumps, file contents) dominate token cost yet rarely need to be replayed verbatim. This strategy truncates only TOOL role message content past max_characters and leaves the user/assistant turns untouched, so the model still sees that a tool ran and a clipped head of its output.

__post_init__()

Reject a non-positive truncation budget.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

def __post_init__(self) -> None:
    """Reject a non-positive truncation budget."""
    if self.max_characters <= 0:
        raise AgentDefinitionError(
            "Tool-result trim budget must be a positive character count"
        )

compact(history, usage, capability) async

Truncate over-budget tool-result content, leaving other roles intact.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Truncate over-budget tool-result content, leaving other roles intact."""
    return tuple(self._trim(message) for message in history)

ProviderManagedCompactionStrategy() dataclass

Bases: ICompactionStrategy

No-op strategy for backends that manage their own context window.

Some providers compact server-side, so the framework must not also trim the transcript. Declaring this strategy makes that hand-off explicit in the spec rather than leaving compaction silently absent.

compact(history, usage, capability) async

Return the history unchanged — the provider owns compaction.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Return the history unchanged — the provider owns compaction."""
    return history

SummarizeOldTurnsCompactionStrategy(model, keep_recent, summary_instruction=DEFAULT_SUMMARY_INSTRUCTION) dataclass

Bases: ICompactionStrategy

Replace older turns with a model-generated summary, keeping recent ones.

The richest compaction: a secondary model call condenses the turns older than keep_recent into one EVIDENCE summary message that precedes the kept tail. This preserves earlier context as a briefing instead of dropping it, at the cost of one extra model round-trip. When the history is already within keep_recent there is nothing older to summarize, so the call is skipped.

__post_init__()

Reject a window that would leave no recent turns to anchor the summary.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

def __post_init__(self) -> None:
    """Reject a window that would leave no recent turns to anchor the summary."""
    if self.keep_recent <= 0:
        raise AgentDefinitionError(
            "Summarize compaction must keep at least one recent message"
        )
    if not self.summary_instruction.strip():
        raise AgentDefinitionError(
            "Summarize compaction instruction cannot be blank"
        )

compact(history, usage, capability) async

Summarize turns older than keep_recent ahead of the recent tail.

Source code in core/spakky-agent/src/spakky/agent/compaction.py

async def compact(
    self,
    history: tuple[ModelMessage, ...],
    usage: ModelUsage,
    capability: ModelCapability,
) -> tuple[ModelMessage, ...]:
    """Summarize turns older than ``keep_recent`` ahead of the recent tail."""
    if len(history) <= self.keep_recent:
        return history
    older = history[: len(history) - self.keep_recent]
    recent = history[len(history) - self.keep_recent :]
    summary = await self.model.complete(self._summary_request(older))
    return (self._summary_message(summary.content), *recent)

Recovery

Action-boundary recovery contracts for durable agent execution.

AgentActionKind

Bases: StrEnum

Recoverable external action classes in an agent execution.

AgentActionBoundaryStage

Bases: StrEnum

Checkpoint side recorded around an action boundary.

AgentResumeAction

Bases: StrEnum

Orchestration action selected from persisted checkpoint evidence.

AgentActionBoundaryCheckpoint(action_id, action_kind, stage, idempotency=Idempotency.UNKNOWN, metadata=dict()) dataclass

Serializable checkpoint recorded before or after one external action.

__post_init__()

Reject checkpoints that cannot be correlated after restart.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

def __post_init__(self) -> None:
    """Reject checkpoints that cannot be correlated after restart."""
    if not self.action_id.strip():
        raise AgentDefinitionError("Agent action boundary id cannot be blank")

before_model_call(action_id, *, idempotency=Idempotency.UNKNOWN, metadata=None) classmethod

Create the checkpoint recorded before a model call is attempted.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def before_model_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency = Idempotency.UNKNOWN,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded before a model call is attempted."""
    return cls._for_action(
        action_id,
        AgentActionKind.MODEL_CALL,
        AgentActionBoundaryStage.BEFORE,
        idempotency=idempotency,
        metadata=metadata,
    )

after_model_call(action_id, *, idempotency=Idempotency.UNKNOWN, metadata=None) classmethod

Create the checkpoint recorded after a model call completes.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def after_model_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency = Idempotency.UNKNOWN,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded after a model call completes."""
    return cls._for_action(
        action_id,
        AgentActionKind.MODEL_CALL,
        AgentActionBoundaryStage.AFTER,
        idempotency=idempotency,
        metadata=metadata,
    )

before_tool_call(action_id, *, idempotency, metadata=None) classmethod

Create the checkpoint recorded before a tool call is attempted.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def before_tool_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded before a tool call is attempted."""
    return cls._for_action(
        action_id,
        AgentActionKind.TOOL_CALL,
        AgentActionBoundaryStage.BEFORE,
        idempotency=idempotency,
        metadata=metadata,
    )

after_tool_call(action_id, *, idempotency, metadata=None) classmethod

Create the checkpoint recorded after a tool call completes.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def after_tool_call(
    cls,
    action_id: str,
    *,
    idempotency: Idempotency,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded after a tool call completes."""
    return cls._for_action(
        action_id,
        AgentActionKind.TOOL_CALL,
        AgentActionBoundaryStage.AFTER,
        idempotency=idempotency,
        metadata=metadata,
    )

before_approval_wait(action_id, *, metadata=None) classmethod

Create the checkpoint recorded before waiting for approval.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def before_approval_wait(
    cls,
    action_id: str,
    *,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded before waiting for approval."""
    return cls._for_action(
        action_id,
        AgentActionKind.APPROVAL_WAIT,
        AgentActionBoundaryStage.BEFORE,
        idempotency=Idempotency.NON_IDEMPOTENT,
        metadata=metadata,
    )

after_approval_wait(action_id, *, metadata=None) classmethod

Create the checkpoint recorded after an approval wait resolves.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

@classmethod
def after_approval_wait(
    cls,
    action_id: str,
    *,
    metadata: JsonObject | None = None,
) -> "AgentActionBoundaryCheckpoint":
    """Create the checkpoint recorded after an approval wait resolves."""
    return cls._for_action(
        action_id,
        AgentActionKind.APPROVAL_WAIT,
        AgentActionBoundaryStage.AFTER,
        idempotency=Idempotency.NON_IDEMPOTENT,
        metadata=metadata,
    )

to_evidence_candidate(*, summary=None)

Represent this checkpoint as append-only evidence.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

def to_evidence_candidate(
    self,
    *,
    summary: str | None = None,
) -> AgentEvidenceCandidate:
    """Represent this checkpoint as append-only evidence."""
    return AgentEvidenceCandidate(
        kind=AgentEvidenceKind.ACTION_BOUNDARY,
        payload={
            "action_id": self.action_id,
            "action_kind": self.action_kind.value,
            "stage": self.stage.value,
            "idempotency": self.idempotency.value,
            "metadata": self.metadata,
        },
        summary=summary,
    )

AgentResumeBoundary(action_id, action_kind, stage, idempotency, evidence_id) dataclass

Last action boundary reconstructed from append-only evidence.

AgentResumePlan(state, action, boundary=None, signals=()) dataclass

Resume decision derived from persisted state and checkpoint evidence.

requires_human_input property

Return whether automatic replay must stop at HITL recovery.

can_resume_automatically property

Return whether orchestration may continue without approval.

plan_agent_resume(state, evidence, signals=())

Restore the next resume action using only persisted state and evidence.

Source code in core/spakky-agent/src/spakky/agent/recovery.py

def plan_agent_resume(
    state: AgentState,
    evidence: Sequence[AgentEvidence],
    signals: Sequence[AgentSignal] = (),
) -> AgentResumePlan:
    """Restore the next resume action using only persisted state and evidence."""
    persisted_signals = tuple(signals)
    if state.status in (
        AgentStatus.COMPLETED,
        AgentStatus.FAILED,
        AgentStatus.CANCELLED,
    ):
        return AgentResumePlan(
            state=state,
            action=AgentResumeAction.NOT_RESUMABLE,
            signals=persisted_signals,
        )

    boundary = _last_action_boundary(evidence)
    if boundary is None:
        return AgentResumePlan(
            state=state,
            action=AgentResumeAction.START,
            signals=persisted_signals,
        )

    if boundary.stage is AgentActionBoundaryStage.AFTER:
        return AgentResumePlan(
            state=replace(state, recovery_marker=boundary.action_id),
            action=AgentResumeAction.SKIP_COMPLETED,
            boundary=boundary,
            signals=persisted_signals,
        )

    if boundary.action_kind is AgentActionKind.APPROVAL_WAIT:
        return AgentResumePlan(
            state=_interrupt_for_human_recovery(state, boundary),
            action=AgentResumeAction.REQUIRE_HITL,
            boundary=boundary,
            signals=persisted_signals,
        )

    resume_action = ToolResumeMetadata(
        idempotency=boundary.idempotency,
    ).action_for_incomplete_boundary()
    if resume_action is ToolResumeAction.RETRY:
        return AgentResumePlan(
            state=replace(state, recovery_marker=boundary.action_id),
            action=AgentResumeAction.RETRY,
            boundary=boundary,
            signals=persisted_signals,
        )
    return AgentResumePlan(
        state=_interrupt_for_human_recovery(state, boundary),
        action=AgentResumeAction.REQUIRE_HITL,
        boundary=boundary,
        signals=persisted_signals,
    )

Approval

Human-in-the-loop approval workflow contracts.

AgentApprovalBoundaryKind

Bases: StrEnum

Action boundaries where orchestration may require human approval.

AgentApprovalPlanAction

Bases: StrEnum

Approval plan outcome before an action boundary is executed.

AgentApprovalRequest(id, agent_state_id, boundary, prompt, risk, action_ref, allowed_decisions=DEFAULT_APPROVAL_DECISIONS, metadata=dict()) dataclass

Approval request materialized at a risky action boundary.

__post_init__()

Reject approval requests that cannot be matched by a signal.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def __post_init__(self) -> None:
    """Reject approval requests that cannot be matched by a signal."""
    _require_non_blank(self.id, "Agent approval id")
    _require_non_blank(self.agent_state_id, "Agent approval state id")
    _require_non_blank(self.prompt, "Agent approval prompt")
    _require_non_blank(self.action_ref, "Agent approval action ref")
    if len(self.allowed_decisions) == 0:
        raise AgentDefinitionError("Agent approval decisions cannot be empty")

from_tool_descriptor(*, approval_id, agent_state_id, descriptor, prompt=None, action_ref=None, call_id=None, metadata=None) classmethod

Build an approval request from a risky tool descriptor.

Source code in core/spakky-agent/src/spakky/agent/approval.py

@classmethod
def from_tool_descriptor(
    cls,
    *,
    approval_id: str,
    agent_state_id: str,
    descriptor: AgentToolDescriptor,
    prompt: str | None = None,
    action_ref: str | None = None,
    call_id: str | None = None,
    metadata: JsonObject | None = None,
) -> "AgentApprovalRequest":
    """Build an approval request from a risky tool descriptor."""
    request_metadata = _tool_approval_metadata(descriptor, call_id, metadata or {})
    return cls(
        id=approval_id,
        agent_state_id=agent_state_id,
        boundary=AgentApprovalBoundaryKind.TOOL_INVOCATION,
        prompt=prompt or f"Approve tool invocation: {descriptor.name}",
        risk=descriptor.metadata.risk,
        action_ref=action_ref or descriptor.identity.key,
        metadata=request_metadata,
    )

to_state(*, agent_type)

Materialize the approval wait as interrupted lifecycle state.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def to_state(self, *, agent_type: str) -> AgentState:
    """Materialize the approval wait as interrupted lifecycle state."""
    _require_non_blank(agent_type, "Agent approval agent type")
    return AgentState(
        id=self.agent_state_id,
        agent_type=agent_type,
        status=AgentStatus.INTERRUPTED,
        transition=AgentStateTransition.WAITING_APPROVAL,
        reason=AgentStateReason.APPROVAL_REQUIRED,
        current_activity=self.prompt,
        metadata={"approval": self.to_metadata()},
    )

to_yield()

Expose this approval request to an inbound adapter stream.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def to_yield(self) -> AgentYield[Approval]:
    """Expose this approval request to an inbound adapter stream."""
    return AgentYield(
        kind=AgentYieldKind.APPROVAL,
        payload=Approval(
            id=self.id,
            prompt=self.prompt,
            allowed_decisions=self.allowed_decisions,
            metadata=self.to_metadata(),
        ),
    )

to_metadata()

Return JSON-compatible metadata for state, yield, and evidence.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def to_metadata(self) -> JsonObject:
    """Return JSON-compatible metadata for state, yield, and evidence."""
    return {
        "id": self.id,
        "agent_state_id": self.agent_state_id,
        "boundary": self.boundary.value,
        "action_ref": self.action_ref,
        "risk_axes": [axis.value for axis in self.risk.axes],
        "allowed_decisions": [
            decision.value for decision in self.allowed_decisions
        ],
        "metadata": self.metadata,
    }

AgentApprovalPlan(action, request=None, state=None, yield_item=None) dataclass

Plan for an action boundary before executing it.

requires_approval property

Return whether orchestration must wait for a HITL decision.

AgentApprovalDecisionOutcome(request_id, decision, status, transition, reason=None, modified_payload=dict(), comment=None) dataclass

Typed result of an approval decision signal.

plan_agent_tool_approval(*, descriptor, approval_id, agent_state_id, agent_type, prompt=None, action_ref=None, call_id=None, metadata=None)

Plan whether a tool invocation should proceed or wait for approval.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def plan_agent_tool_approval(
    *,
    descriptor: AgentToolDescriptor,
    approval_id: str,
    agent_state_id: str,
    agent_type: str,
    prompt: str | None = None,
    action_ref: str | None = None,
    call_id: str | None = None,
    metadata: JsonObject | None = None,
) -> AgentApprovalPlan:
    """Plan whether a tool invocation should proceed or wait for approval."""
    if not descriptor.metadata.requires_approval_candidate:
        return AgentApprovalPlan(action=AgentApprovalPlanAction.PROCEED)

    request = AgentApprovalRequest.from_tool_descriptor(
        approval_id=approval_id,
        agent_state_id=agent_state_id,
        descriptor=descriptor,
        prompt=prompt,
        action_ref=action_ref,
        call_id=call_id,
        metadata=metadata,
    )
    return AgentApprovalPlan(
        action=AgentApprovalPlanAction.WAIT_FOR_APPROVAL,
        request=request,
        state=request.to_state(agent_type=agent_type),
        yield_item=request.to_yield(),
    )

parse_agent_approval_decision_signal(signal, *, request=None)

Parse an approval decision signal into a typed workflow outcome.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def parse_agent_approval_decision_signal(
    signal: AgentSignal,
    *,
    request: AgentApprovalRequest | None = None,
) -> AgentApprovalDecisionOutcome:
    """Parse an approval decision signal into a typed workflow outcome."""
    if signal.kind != AgentSignalKind.APPROVAL_DECISION:
        raise AgentDefinitionError(
            "Agent approval signal kind must be approval_decision"
        )

    request_id = _require_payload_text(signal.payload, "request_id")
    if request is not None and request_id != request.id:
        raise AgentDefinitionError("Agent approval signal request id does not match")
    if request is not None and signal.agent_state_id != request.agent_state_id:
        raise AgentDefinitionError("Agent approval signal state id does not match")

    decision = _parse_approval_decision(
        _require_payload_text(signal.payload, "decision")
    )
    modified_payload = _payload_object(signal.payload.get("modified_payload", {}))
    comment = _optional_payload_text(signal.payload, "comment")
    status, transition, reason = _decision_state_target(decision)

    return AgentApprovalDecisionOutcome(
        request_id=request_id,
        decision=decision,
        status=status,
        transition=transition,
        reason=reason,
        modified_payload=modified_payload,
        comment=comment,
    )

materialize_agent_approval_decision_state(current, outcome)

Apply a typed approval decision outcome to an existing state snapshot.

Source code in core/spakky-agent/src/spakky/agent/approval.py

def materialize_agent_approval_decision_state(
    current: AgentState,
    outcome: AgentApprovalDecisionOutcome,
) -> AgentState:
    """Apply a typed approval decision outcome to an existing state snapshot."""
    return AgentState(
        id=current.id,
        agent_type=current.agent_type,
        status=outcome.status,
        transition=outcome.transition,
        reason=outcome.reason,
        current_activity=_activity_for_decision(outcome),
        input_ref=current.input_ref,
        output_ref=current.output_ref,
        pending_signal_count=current.pending_signal_count,
        last_event_cursor=current.last_event_cursor,
        recovery_marker=current.recovery_marker,
        metadata={
            **current.metadata,
            "approval_decision": {
                "request_id": outcome.request_id,
                "decision": outcome.decision.value,
                "modified_payload": outcome.modified_payload,
                "comment": outcome.comment,
            },
        },
        created_at=current.created_at,
        updated_at=current.updated_at,
    )

Cancellation

Cancellation lifecycle and cleanup contracts for agent execution.

AgentCancellationCleanupCallable = Callable[[AgentCancellationRequest], Awaitable[AgentCancellationCleanupResult]]

Async hook shape invoked for model stream/tool/delegate cancellation cleanup.

AgentCancellationTargetKind

Bases: StrEnum

Running execution target that can receive a cancellation cleanup hook.

AgentCancellationCleanupStatus

Bases: StrEnum

Outcome of one cancellation cleanup hook.

AgentCancellationRequest(state_id, signal_id, target_kind, target_ref, reason=None, requested_by=None, metadata=dict()) dataclass

Cancellation request passed to a model stream, tool, or delegate hook.

__post_init__()

Reject requests that cannot be traced back to state/signal/target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject requests that cannot be traced back to state/signal/target."""
    _require_non_blank(self.state_id, "Agent cancellation state id")
    _require_non_blank(self.signal_id, "Agent cancellation signal id")
    _require_non_blank(self.target_ref, "Agent cancellation target ref")
    _require_optional_non_blank(self.reason, "Agent cancellation reason")
    _require_optional_non_blank(self.requested_by, "Agent cancellation requester")

from_signal(*, state, signal, target_kind, target_ref, metadata=None) classmethod

Build a hook request from the durable CANCEL signal payload.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def from_signal(
    cls,
    *,
    state: AgentState,
    signal: AgentSignal,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    metadata: JsonObject | None = None,
) -> "AgentCancellationRequest":
    """Build a hook request from the durable CANCEL signal payload."""
    _require_cancel_signal(signal)
    _require_signal_matches_state(state, signal)
    return cls(
        state_id=state.id,
        signal_id=signal.id,
        target_kind=target_kind,
        target_ref=target_ref,
        reason=_optional_string(signal.payload, "reason"),
        requested_by=_optional_string(signal.payload, "requested_by"),
        metadata={} if metadata is None else metadata,
    )

to_payload()

Serialize the request into append-only evidence metadata.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_payload(self) -> JsonObject:
    """Serialize the request into append-only evidence metadata."""
    return {
        "state_id": self.state_id,
        "signal_id": self.signal_id,
        "target_kind": self.target_kind.value,
        "target_ref": self.target_ref,
        "reason": self.reason,
        "requested_by": self.requested_by,
        "metadata": self.metadata,
    }

AgentCancellationCleanupResult(target_kind, target_ref, status, reason=None, error_code=None, message=None, metadata=dict()) dataclass

Result returned by one cancellation cleanup hook.

failed_cleanup property

Return whether this hook prevents a clean CANCELLED terminal state.

__post_init__()

Reject outcomes that cannot be correlated with the cleanup target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject outcomes that cannot be correlated with the cleanup target."""
    _require_non_blank(self.target_ref, "Agent cancellation cleanup target ref")
    _require_optional_non_blank(self.reason, "Agent cancellation cleanup reason")
    _require_optional_non_blank(
        self.error_code,
        "Agent cancellation cleanup error code",
    )
    _require_optional_non_blank(self.message, "Agent cancellation cleanup message")

succeeded(*, target_kind, target_ref, reason=None, metadata=None) classmethod

Record a cleanup hook that released its target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def succeeded(
    cls,
    *,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    reason: str | None = None,
    metadata: JsonObject | None = None,
) -> "AgentCancellationCleanupResult":
    """Record a cleanup hook that released its target."""
    return cls(
        target_kind=target_kind,
        target_ref=target_ref,
        status=AgentCancellationCleanupStatus.SUCCEEDED,
        reason=reason,
        metadata={} if metadata is None else metadata,
    )

failed(*, target_kind, target_ref, error_code, message, metadata=None) classmethod

Record a cleanup hook that could not release its target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def failed(
    cls,
    *,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    error_code: str,
    message: str,
    metadata: JsonObject | None = None,
) -> "AgentCancellationCleanupResult":
    """Record a cleanup hook that could not release its target."""
    return cls(
        target_kind=target_kind,
        target_ref=target_ref,
        status=AgentCancellationCleanupStatus.FAILED,
        error_code=error_code,
        message=message,
        metadata={} if metadata is None else metadata,
    )

skipped(*, target_kind, target_ref, reason, metadata=None) classmethod

Record a cleanup target that was already inactive or unavailable.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

@classmethod
def skipped(
    cls,
    *,
    target_kind: AgentCancellationTargetKind,
    target_ref: str,
    reason: str,
    metadata: JsonObject | None = None,
) -> "AgentCancellationCleanupResult":
    """Record a cleanup target that was already inactive or unavailable."""
    return cls(
        target_kind=target_kind,
        target_ref=target_ref,
        status=AgentCancellationCleanupStatus.SKIPPED,
        reason=reason,
        metadata={} if metadata is None else metadata,
    )

to_payload()

Serialize this hook outcome into evidence/state metadata.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_payload(self) -> JsonObject:
    """Serialize this hook outcome into evidence/state metadata."""
    return {
        "target_kind": self.target_kind.value,
        "target_ref": self.target_ref,
        "status": self.status.value,
        "reason": self.reason,
        "error_code": self.error_code,
        "message": self.message,
        "metadata": self.metadata,
    }

AgentCancellationCleanupTask(target_kind, target_ref, cleanup, metadata=dict()) dataclass

One cleanup hook registered for a running cancellation target.

__post_init__()

Reject hooks that cannot be identified in cleanup evidence.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject hooks that cannot be identified in cleanup evidence."""
    _require_non_blank(self.target_ref, "Agent cancellation cleanup target ref")

run(*, state, signal) async

Invoke the hook and verify that it reports the same target.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

async def run(
    self,
    *,
    state: AgentState,
    signal: AgentSignal,
) -> AgentCancellationCleanupResult:
    """Invoke the hook and verify that it reports the same target."""
    request = AgentCancellationRequest.from_signal(
        state=state,
        signal=signal,
        target_kind=self.target_kind,
        target_ref=self.target_ref,
        metadata=self.metadata,
    )
    result = await self.cleanup(request)
    if (
        result.target_kind is not self.target_kind
        or result.target_ref != self.target_ref
    ):
        raise AgentDefinitionError(
            "Agent cancellation cleanup result target must match the task"
        )
    return result

AgentCancellationCleanupReport(state_id, signal_id, outcomes=()) dataclass

Aggregate cleanup evidence for one CANCEL signal.

cleanup_succeeded property

Return whether all cleanup hooks completed without failure.

failed_outcomes property

Return the hook outcomes that force FAILED terminal state.

__post_init__()

Reject reports that cannot be attached to state and signal evidence.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def __post_init__(self) -> None:
    """Reject reports that cannot be attached to state and signal evidence."""
    _require_non_blank(self.state_id, "Agent cancellation report state id")
    _require_non_blank(self.signal_id, "Agent cancellation report signal id")

to_payload()

Serialize the report for state metadata and evidence payloads.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_payload(self) -> JsonObject:
    """Serialize the report for state metadata and evidence payloads."""
    return {
        "state_id": self.state_id,
        "signal_id": self.signal_id,
        "cleanup_succeeded": self.cleanup_succeeded,
        "outcomes": tuple(outcome.to_payload() for outcome in self.outcomes),
    }

to_evidence_candidate(*, summary=None)

Represent cancellation cleanup as append-only evidence.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def to_evidence_candidate(
    self,
    *,
    summary: str | None = None,
) -> AgentEvidenceCandidate:
    """Represent cancellation cleanup as append-only evidence."""
    return AgentEvidenceCandidate(
        kind=AgentEvidenceKind.CANCELLATION,
        payload=self.to_payload(),
        summary=summary,
    )

run_agent_cancellation_cleanup(*, state, signal, tasks) async

Invoke model stream/tool/delegate cleanup hooks for a CANCEL signal.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

async def run_agent_cancellation_cleanup(
    *,
    state: AgentState,
    signal: AgentSignal,
    tasks: Sequence[AgentCancellationCleanupTask],
) -> AgentCancellationCleanupReport:
    """Invoke model stream/tool/delegate cleanup hooks for a CANCEL signal."""
    _require_cancel_signal(signal)
    outcomes: list[AgentCancellationCleanupResult] = []
    for task in tasks:
        outcomes.append(await task.run(state=state, signal=signal))
    return AgentCancellationCleanupReport(
        state_id=state.id,
        signal_id=signal.id,
        outcomes=tuple(outcomes),
    )

begin_agent_cancellation(state, signal)

Materialize receipt of a CANCEL signal as CANCELLING state.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def begin_agent_cancellation(
    state: AgentState,
    signal: AgentSignal,
) -> AgentState:
    """Materialize receipt of a CANCEL signal as CANCELLING state."""
    _require_cancel_signal(signal)
    _require_signal_matches_state(state, signal)
    return replace(
        state,
        status=AgentStatus.CANCELLING,
        transition=AgentStateTransition.CANCELLING,
        reason=AgentStateReason.CANCELLATION_REQUESTED,
        metadata={
            **state.metadata,
            "cancellation_signal_id": signal.id,
            "cancellation_reason": _optional_string(signal.payload, "reason"),
            "cancellation_requested_by": _optional_string(
                signal.payload,
                "requested_by",
            ),
        },
    )

complete_agent_cancellation(state, report)

Resolve CANCELLING state into CANCELLED or FAILED cleanup outcome.

Source code in core/spakky-agent/src/spakky/agent/cancellation.py

def complete_agent_cancellation(
    state: AgentState,
    report: AgentCancellationCleanupReport,
) -> AgentState:
    """Resolve CANCELLING state into CANCELLED or FAILED cleanup outcome."""
    if state.id != report.state_id:
        raise AgentDefinitionError("Agent cancellation report state id mismatch")
    metadata = {**state.metadata, "cancellation_cleanup": report.to_payload()}
    if report.cleanup_succeeded:
        return replace(
            state,
            status=AgentStatus.CANCELLED,
            transition=AgentStateTransition.CANCELLED,
            reason=AgentStateReason.CANCELLATION_REQUESTED,
            metadata=metadata,
        )
    return replace(
        state,
        status=AgentStatus.FAILED,
        transition=AgentStateTransition.FAILED,
        reason=AgentStateReason.CANCELLATION_CLEANUP_FAILED,
        metadata=metadata,
    )

Delegation

Agent-to-agent delegation contracts.

DelegationReturnPolicy

Bases: StrEnum

How a child agent result should be projected back to the parent.

AgentDelegateTarget(agent_type, agent_name=None, metadata=dict()) dataclass

First-class delegate target represented by another @Agent component.

__post_init__()

Reject delegate targets that cannot identify an agent component.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegate targets that cannot identify an agent component."""
    if not self.agent_type.strip():
        raise AgentDefinitionError("Delegate target agent type cannot be blank")
    if self.agent_name is not None and not self.agent_name.strip():
        raise AgentDefinitionError("Delegate target agent name cannot be blank")

DelegationBudget(max_steps=None, max_tokens=None, timeout_seconds=None, deadline_at=None, metadata=dict()) dataclass

Budget metadata attached to a delegation packet.

__post_init__()

Reject delegation budgets that cannot be enforced consistently.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegation budgets that cannot be enforced consistently."""
    if self.max_steps is not None and self.max_steps <= 0:
        raise AgentDefinitionError("Delegation max steps must be positive")
    if self.max_tokens is not None and self.max_tokens <= 0:
        raise AgentDefinitionError("Delegation max tokens must be positive")
    if self.timeout_seconds is not None and self.timeout_seconds <= 0:
        raise AgentDefinitionError("Delegation timeout must be positive")

DelegationContextSlice(summary=None, evidence_refs=(), manifest_ref=None, metadata=dict()) dataclass

Minimal parent context projected for a child agent.

DelegationExpectedOutput(description=None, schema=dict(), metadata=dict()) dataclass

Expected child output description and optional JSON schema.

DelegationPacket(id, parent_agent_state_id, target, task, context=DelegationContextSlice(), constraints=(), expected_output=DelegationExpectedOutput(), budget=DelegationBudget(), allowed_capabilities=(), return_policy=DelegationReturnPolicy.SUMMARY_AND_EVIDENCE, metadata=dict()) dataclass

Task packet passed from a parent agent to a delegate agent.

__post_init__()

Reject delegation packets without parent linkage or task identity.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegation packets without parent linkage or task identity."""
    if not self.id.strip():
        raise AgentDefinitionError("Delegation id cannot be blank")
    if not self.parent_agent_state_id.strip():
        raise AgentDefinitionError("Delegation parent state id cannot be blank")

DelegationResult(id, packet_id, target, summary, output=None, evidence_refs=(), metadata=dict(), created_at=None) dataclass

Child agent result projected back to the parent execution.

__post_init__()

Reject delegated results that cannot be linked to a packet.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegated results that cannot be linked to a packet."""
    if not self.id.strip():
        raise AgentDefinitionError("Delegation result id cannot be blank")
    if not self.packet_id.strip():
        raise AgentDefinitionError("Delegation result packet id cannot be blank")
    if not self.summary.strip():
        raise AgentDefinitionError("Delegation result summary cannot be blank")

to_parent_evidence(*, evidence_id, parent_agent_state_id)

Represent a delegated result as append-only parent evidence.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def to_parent_evidence(
    self,
    *,
    evidence_id: str,
    parent_agent_state_id: str,
) -> AgentEvidence:
    """Represent a delegated result as append-only parent evidence."""
    payload: dict[str, JsonValue] = {
        "delegation_id": self.id,
        "packet_id": self.packet_id,
        "target_agent_type": self.target.agent_type,
        "evidence_refs": self.evidence_refs,
        "metadata": self.metadata,
    }
    if self.target.agent_name is not None:
        payload["target_agent_name"] = self.target.agent_name
    if self.output is not None:
        payload["output"] = self.output
    return AgentEvidence(
        id=evidence_id,
        agent_state_id=parent_agent_state_id,
        kind=AgentEvidenceKind.DELEGATION,
        payload=payload,
        summary=self.summary,
        created_at=self.created_at,
    )

to_parent_yield(*, evidence_id, parent_agent_state_id)

Expose the delegated result on the parent's AgentYield stream.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def to_parent_yield(
    self,
    *,
    evidence_id: str,
    parent_agent_state_id: str,
) -> AgentYield[Evidence]:
    """Expose the delegated result on the parent's AgentYield stream."""
    evidence = self.to_parent_evidence(
        evidence_id=evidence_id,
        parent_agent_state_id=parent_agent_state_id,
    )
    return AgentYield(
        kind=AgentYieldKind.EVIDENCE,
        payload=Evidence(
            evidence=evidence,
            metadata={"delegation_id": self.id, "packet_id": self.packet_id},
        ),
    )

DelegationToolResult(summary, output=None, events=(), metadata=dict()) dataclass

Model-facing result plus child neutral events from a teammate call.

__post_init__()

Reject delegated tool results without a model-facing summary.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def __post_init__(self) -> None:
    """Reject delegated tool results without a model-facing summary."""
    if not self.summary.strip():
        raise AgentDefinitionError("Delegation tool result summary cannot be blank")

from_result(result, *, events=()) classmethod

Build a model-facing tool result from a delegation result object.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

@classmethod
def from_result(
    cls,
    result: DelegationResult,
    *,
    events: tuple[AgentEvent, ...] = (),
) -> "DelegationToolResult":
    """Build a model-facing tool result from a delegation result object."""
    return cls(
        summary=result.summary,
        output=result.output,
        events=events,
        metadata={
            "delegation_id": result.id,
            "packet_id": result.packet_id,
            "target_agent_type": result.target.agent_type,
            **result.metadata,
        },
    )

IAgentDelegate

Bases: ABC

Execution hook that runs a delegation packet against a delegate target.

delegate(packet) abstractmethod

Execute delegation without prescribing spawn topology or transport.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

@abstractmethod
def delegate(
    self,
    packet: DelegationPacket,
) -> AsyncGenerator[AgentYield[DelegationResult], None]:
    """Execute delegation without prescribing spawn topology or transport."""

delegate_tool_result(packet) async

Execute delegation and collect the terminal result for a tool call.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

async def delegate_tool_result(
    self,
    packet: DelegationPacket,
) -> DelegationToolResult:
    """Execute delegation and collect the terminal result for a tool call."""
    terminal: DelegationResult | None = None
    async for item in self.delegate(packet):
        if isinstance(item.payload, DelegationResult):
            terminal = item.payload
    if terminal is None:
        raise AgentToolDispatchError("Agent delegate produced no delegation result")
    return DelegationToolResult.from_result(terminal)

build_teammate_tool_descriptors(owner, teammates)

Build model-callable delegation tools from an @Agent teammate spec.

Source code in core/spakky-agent/src/spakky/agent/delegation.py

def build_teammate_tool_descriptors(
    owner: type[object],
    teammates,
) -> tuple[AgentToolDescriptor, ...]:
    """Build model-callable delegation tools from an ``@Agent`` teammate spec."""
    return tuple(
        _build_teammate_tool_descriptor(owner, teammate) for teammate in teammates
    )

Safety

Deterministic sensitive-data contracts for agent boundaries.

DataSensitivity

Bases: StrEnum

Canonical sensitivity classes carried by descriptors.

PII

Bases: StrEnum

PII categories that can be declared with typing.Annotated.

MaskingPolicy

Bases: StrEnum

Deterministic text masking strategies for sensitive values.

RedactionPolicy

Bases: StrEnum

Boundary action used when a value must not be exposed.

StreamingGuardFailureMode

Bases: StrEnum

Final audit behavior when a streaming guard missed a raw candidate.

StreamingRedactionAuditStatus

Bases: StrEnum

Final aggregate streaming redaction audit status.

ContextExposurePolicy(include_pii_values=False, include_sensitive_values=False, include_sensitive_schema_metadata=False, include_sensitive_context_metadata=False) dataclass

Policy for LLM-facing context and schema metadata exposure.

can_expose_value(field)

Return whether a sensitive value may cross the model boundary.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def can_expose_value(self, field: "SensitiveField | SecretField") -> bool:
    """Return whether a sensitive value may cross the model boundary."""
    if isinstance(field, SecretField):
        return False
    if field.sensitivity == DataSensitivity.PII:
        return self.include_pii_values
    if field.sensitivity in (DataSensitivity.CONFIDENTIAL, DataSensitivity.SECRET):
        return self.include_sensitive_values
    return True

EvidenceExposurePolicy(include_pii_values=False, include_sensitive_values=False, include_sensitive_metadata=True) dataclass

Policy for evidence payload exposure before append-only capture.

can_expose_value(field)

Return whether a sensitive value may be stored in evidence payloads.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def can_expose_value(self, field: "SensitiveField | SecretField") -> bool:
    """Return whether a sensitive value may be stored in evidence payloads."""
    if isinstance(field, SecretField):
        return False
    if field.sensitivity == DataSensitivity.PII:
        return self.include_pii_values
    if field.sensitivity in (DataSensitivity.CONFIDENTIAL, DataSensitivity.SECRET):
        return self.include_sensitive_values
    return True

CredentialRef(id, provider=None) dataclass

Reference to a credential outside LLM-facing context.

__post_init__()

Reject blank credential references.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank credential references."""
    _require_non_blank(self.id, "Credential reference id")
    if self.provider is not None:
        _require_non_blank(self.provider, "Credential provider")

SecretRef(id, credential=None) dataclass

Opaque reference to a secret value stored outside model context.

__post_init__()

Reject blank secret references.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank secret references."""
    _require_non_blank(self.id, "Secret reference id")

SensitiveField(category, masking=MaskingPolicy.REDACT, redaction=RedactionPolicy.REDACT, label=None, metadata=dict()) dataclass

typing.Annotated metadata for deterministic sensitive-field handling.

sensitivity property

Return the normalized sensitivity class.

category_name property

Return the stable public category name.

__post_init__()

Reject blank sensitive field labels.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank sensitive field labels."""
    if self.label is not None:
        _require_non_blank(self.label, "Sensitive field label")

guard_text(value)

Return deterministic model/evidence-safe text for this field.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def guard_text(self, value: str) -> str:
    """Return deterministic model/evidence-safe text for this field."""
    return _mask_text(value, self.masking, self.category_name)

to_metadata()

Serialize marker metadata without including the sensitive value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_metadata(self) -> JsonObject:
    """Serialize marker metadata without including the sensitive value."""
    payload: dict[str, JsonValue] = {
        "kind": "sensitive",
        "category": self.category_name,
        "sensitivity": self.sensitivity.value,
        "masking": self.masking.value,
        "redaction": self.redaction.value,
    }
    if self.label is not None:
        payload["label"] = self.label
    if self.metadata:
        payload["metadata"] = self.metadata
    return payload

SecretField(redaction=RedactionPolicy.REFERENCE_ONLY, label=None, metadata=dict()) dataclass

typing.Annotated metadata for values that must never be model text.

sensitivity property

Return the normalized sensitivity class.

__post_init__()

Reject blank secret field labels.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank secret field labels."""
    if self.label is not None:
        _require_non_blank(self.label, "Secret field label")

guard_text(value)

Return deterministic replacement text for a secret value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def guard_text(self, value: str) -> str:
    """Return deterministic replacement text for a secret value."""
    return SECRET_VALUE

to_metadata()

Serialize marker metadata without including the secret value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_metadata(self) -> JsonObject:
    """Serialize marker metadata without including the secret value."""
    payload: dict[str, JsonValue] = {
        "kind": "secret",
        "sensitivity": DataSensitivity.SECRET.value,
        "redaction": self.redaction.value,
    }
    if self.label is not None:
        payload["label"] = self.label
    if self.metadata:
        payload["metadata"] = self.metadata
    return payload

SensitiveFieldDescriptor(path, field) dataclass

Path-bound sensitive metadata extracted from Annotated types.

to_metadata()

Serialize descriptor metadata without leaking the field value.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_metadata(self) -> JsonObject:
    """Serialize descriptor metadata without leaking the field value."""
    return {
        "path": list(self.path),
        "field": self.field.to_metadata(),
    }

StreamingSensitivePattern(name, pattern, replacement=REDACTED_VALUE, metadata=dict()) dataclass

Caller-supplied deterministic pattern used by streaming redaction.

__post_init__()

Reject blank names and invalid pattern syntax before streaming starts.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject blank names and invalid pattern syntax before streaming starts."""
    _require_non_blank(self.name, "Streaming sensitive pattern name")
    _require_non_blank(self.pattern, "Streaming sensitive pattern")
    try:
        re.compile(self.pattern)
    except re.error as e:
        raise AgentDefinitionError("Streaming sensitive pattern is invalid") from e

redact(value)

Return redacted text and the number of replacements applied.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def redact(self, value: str) -> tuple[str, int]:
    """Return redacted text and the number of replacements applied."""
    compiled = re.compile(self.pattern)
    return compiled.subn(self.replacement, value)

find_matches(value)

Return sanitized match locations without exposing raw text.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def find_matches(self, value: str) -> tuple["StreamingRedactionMatch", ...]:
    """Return sanitized match locations without exposing raw text."""
    compiled = re.compile(self.pattern)
    return tuple(
        StreamingRedactionMatch(
            pattern_name=self.name,
            start=match.start(),
            end=match.end(),
            metadata=self.metadata,
        )
        for match in compiled.finditer(value)
    )

StreamingRedactionPolicy(patterns, buffer_size=64, emit_chunk_size=None, failure_mode=StreamingGuardFailureMode.RAISE) dataclass

Bounded buffering policy balancing stream latency and redaction correctness.

__post_init__()

Reject policies that would make the guard unbounded or silent.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __post_init__(self) -> None:
    """Reject policies that would make the guard unbounded or silent."""
    if self.buffer_size <= 0:
        raise AgentDefinitionError(
            "Streaming redaction buffer size must be positive"
        )
    if self.emit_chunk_size is not None and self.emit_chunk_size <= 0:
        raise AgentDefinitionError(
            "Streaming redaction emit chunk size must be positive"
        )
    if len(self.patterns) == 0:
        raise AgentDefinitionError("Streaming redaction patterns cannot be empty")

StreamingRedactionMatch(pattern_name, start, end, metadata=dict()) dataclass

Sanitized final-audit match location for a missed redaction candidate.

to_payload()

Serialize a match without the sensitive value itself.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_payload(self) -> JsonObject:
    """Serialize a match without the sensitive value itself."""
    payload: dict[str, JsonValue] = {
        "pattern_name": self.pattern_name,
        "start": self.start,
        "end": self.end,
    }
    if self.metadata:
        payload["metadata"] = self.metadata
    return payload

StreamingRedactionAudit(status, detected_count, redacted_count, missed_matches, buffer_size, emitted_char_count, original_char_count) dataclass

Final aggregate audit for a bounded streaming redaction session.

missed_count property

Return the number of raw candidates still present after streaming.

to_evidence_payload()

Serialize audit evidence without raw streamed content.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_evidence_payload(self) -> JsonObject:
    """Serialize audit evidence without raw streamed content."""
    return {
        "kind": "streaming_redaction_audit",
        "status": self.status.value,
        "detected_count": self.detected_count,
        "redacted_count": self.redacted_count,
        "missed_count": self.missed_count,
        "missed_matches": tuple(
            match.to_payload() for match in self.missed_matches
        ),
        "buffer_size": self.buffer_size,
        "emitted_char_count": self.emitted_char_count,
        "original_char_count": self.original_char_count,
    }

to_error_payload()

Serialize a typed error payload for stream consumers.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def to_error_payload(self) -> JsonObject:
    """Serialize a typed error payload for stream consumers."""
    return {
        "code": "streaming_redaction_audit_failed",
        "message": "Streaming redaction final audit detected unmasked candidates",
        "retryable": False,
        "metadata": self.to_evidence_payload(),
    }

StreamingRedactionResult(chunks, audit=None, error=None) dataclass

Output produced by a bounded streaming redaction step.

StreamingRedactionSession(policy)

Stateful bounded redactor for model token streams.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def __init__(self, policy: StreamingRedactionPolicy) -> None:
    self._policy = policy
    self._patterns = tuple(policy.patterns)
    self._buffer = ""
    self._original_chunks: list[str] = []
    self._emitted_chunks: list[str] = []
    self._redacted_count = 0
    self._closed = False

push(chunk)

Redact one token chunk and emit only the safe bounded prefix.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def push(self, chunk: str) -> StreamingRedactionResult:
    """Redact one token chunk and emit only the safe bounded prefix."""
    self._ensure_open()
    if chunk == "":
        return StreamingRedactionResult(chunks=())
    self._original_chunks.append(chunk)
    self._buffer = self._redact_text(f"{self._buffer}{chunk}")
    chunks = self._emit_safe_prefix()
    return StreamingRedactionResult(chunks=chunks)

finish()

Flush the remaining buffer and run the mandatory final audit.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def finish(self) -> StreamingRedactionResult:
    """Flush the remaining buffer and run the mandatory final audit."""
    self._ensure_open()
    self._closed = True
    redacted_buffer = self._redact_text(self._buffer)
    chunks = self._split_for_emit(redacted_buffer)
    self._buffer = ""
    self._emitted_chunks.extend(chunks)
    audit = self._audit()
    if audit.status == StreamingRedactionAuditStatus.FAILED:
        if self._policy.failure_mode == StreamingGuardFailureMode.RAISE:
            raise AgentOutputGuardError(
                "Streaming redaction final audit detected unmasked candidates"
            )
        return StreamingRedactionResult(
            chunks=chunks,
            audit=audit,
            error=audit.to_error_payload(),
        )
    return StreamingRedactionResult(chunks=chunks, audit=audit)

guard_json_value(value, sensitive_fields, policy)

Redact JSON-compatible values according to path-bound descriptors.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def guard_json_value(
    value: JsonValue,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: ContextExposurePolicy | EvidenceExposurePolicy,
) -> JsonValue:
    """Redact JSON-compatible values according to path-bound descriptors."""
    result = value
    for descriptor in sensitive_fields:
        result = _guard_json_path(result, descriptor.path, descriptor.field, policy)
    return result

schema_with_sensitive_metadata(schema, sensitive_fields, policy)

Return a JSON schema copy with policy-approved sensitivity extensions.

Source code in core/spakky-agent/src/spakky/agent/safety.py

def schema_with_sensitive_metadata(
    schema: Mapping[str, JsonValue],
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: ContextExposurePolicy,
) -> JsonObject:
    """Return a JSON schema copy with policy-approved sensitivity extensions."""
    if not policy.include_sensitive_schema_metadata or not sensitive_fields:
        return schema
    result = _copy_json_object(schema)
    for descriptor in sensitive_fields:
        _attach_schema_extension(result, descriptor.path, descriptor.to_metadata())
    return result

Tooling

Agent tool descriptor discovery contracts.

Idempotency

Bases: StrEnum

Action idempotency declared by a tool.

ToolResumeAction

Bases: StrEnum

Resume action allowed by stored tool idempotency metadata.

DataAccess

Bases: StrEnum

Data access level declared by a tool.

Externality

Bases: StrEnum

External side-effect boundary declared by a tool.

EvidenceCapture

Bases: StrEnum

Evidence capture strategy for tool results.

ToolApprovalRequirement

Bases: StrEnum

Human approval requirement at the tool boundary.

ToolRiskAxis

Bases: StrEnum

Derived risk axes exposed for policy and UI decisions.

ToolPermission(name) dataclass

Typed permission marker attached to a tool descriptor.

__post_init__()

Reject permission names that cannot be matched deterministically.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject permission names that cannot be matched deterministically."""
    if not self.name.strip():
        raise AgentDefinitionError("Agent tool permission name cannot be blank")

ToolEffects(data_access=DataAccess.NONE, externality=Externality.LOCAL, destructive=False, network=False) dataclass

Typed effect metadata used to derive display risk outside core.

read_only() classmethod

Declare a local read-only tool.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def read_only(cls) -> "ToolEffects":
    """Declare a local read-only tool."""
    return cls(data_access=DataAccess.READ, externality=Externality.LOCAL)

write_state() classmethod

Declare a local state-writing tool.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def write_state(cls) -> "ToolEffects":
    """Declare a local state-writing tool."""
    return cls(data_access=DataAccess.WRITE, externality=Externality.LOCAL)

external_side_effect() classmethod

Declare a tool that crosses an external side-effect boundary.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def external_side_effect(cls) -> "ToolEffects":
    """Declare a tool that crosses an external side-effect boundary."""
    return cls(
        data_access=DataAccess.READ_WRITE,
        externality=Externality.EXTERNAL,
        network=True,
    )

destructive_action() classmethod

Declare a tool that may irreversibly mutate local or external state.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def destructive_action(cls) -> "ToolEffects":
    """Declare a tool that may irreversibly mutate local or external state."""
    return cls(
        data_access=DataAccess.WRITE,
        externality=Externality.EXTERNAL,
        destructive=True,
    )

TimeoutPolicy(seconds=None) dataclass

Optional timeout boundary for a tool invocation.

__post_init__()

Reject non-positive timeout policies at definition time.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject non-positive timeout policies at definition time."""
    if self.seconds is not None and self.seconds <= 0:
        raise AgentDefinitionError("Agent tool timeout must be positive")

ResultBudget(max_bytes=None) dataclass

Optional result-size budget for model-facing tool output.

__post_init__()

Reject result budgets that cannot constrain output.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject result budgets that cannot constrain output."""
    if self.max_bytes is not None and self.max_bytes <= 0:
        raise AgentDefinitionError("Agent tool result budget must be positive")

ToolResumeMetadata(idempotency=Idempotency.UNKNOWN) dataclass

Stored idempotency metadata used when resuming an incomplete action.

from_metadata(metadata) classmethod

Build resume metadata from a tool descriptor metadata object.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def from_metadata(cls, metadata: "AgentToolMetadata") -> "ToolResumeMetadata":
    """Build resume metadata from a tool descriptor metadata object."""
    return cls(idempotency=metadata.idempotency)

action_for_completed_boundary()

Return the resume action for an already completed action boundary.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def action_for_completed_boundary(self) -> ToolResumeAction:
    """Return the resume action for an already completed action boundary."""
    return ToolResumeAction.SKIP_COMPLETED

action_for_incomplete_boundary()

Return the resume action for an incomplete action boundary.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def action_for_incomplete_boundary(self) -> ToolResumeAction:
    """Return the resume action for an incomplete action boundary."""
    if self.idempotency in (
        Idempotency.IDEMPOTENT,
        Idempotency.CONDITIONALLY_IDEMPOTENT,
    ):
        return ToolResumeAction.RETRY
    return ToolResumeAction.REQUIRE_APPROVAL

AgentToolSchemaHandle(name, input_schema_name, output_schema_name, input_schema=dict(), output_schema=dict(), input_sensitive_fields=(), output_sensitive_fields=()) dataclass

Stable schema names and generated JSON schemas owned by a descriptor.

__post_init__()

Reject blank schema handles.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject blank schema handles."""
    _require_non_blank(self.name, "Agent tool schema name")
    _require_non_blank(self.input_schema_name, "Agent tool input schema name")
    _require_non_blank(self.output_schema_name, "Agent tool output schema name")

input_schema_for(policy=None)

Return model-facing input schema under the requested exposure policy.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def input_schema_for(
    self,
    policy: ContextExposurePolicy | None = None,
) -> JsonObject:
    """Return model-facing input schema under the requested exposure policy."""
    return schema_with_sensitive_metadata(
        self.input_schema,
        self.input_sensitive_fields,
        policy or ContextExposurePolicy(),
    )

output_schema_for(policy=None)

Return model-facing output schema under the requested exposure policy.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def output_schema_for(
    self,
    policy: ContextExposurePolicy | None = None,
) -> JsonObject:
    """Return model-facing output schema under the requested exposure policy."""
    return schema_with_sensitive_metadata(
        self.output_schema,
        self.output_sensitive_fields,
        policy or ContextExposurePolicy(),
    )

AgentToolIdentity(owner_module, owner_qualname, name) dataclass

Descriptor identity independent from free-form model output text.

key property

Return a stable key for maps, logs, and evidence metadata.

__post_init__()

Reject identity parts that cannot be serialized or matched.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject identity parts that cannot be serialized or matched."""
    _require_non_blank(self.owner_module, "Agent tool owner module")
    _require_non_blank(self.owner_qualname, "Agent tool owner qualname")
    _require_non_blank(self.name, "Agent tool name")

AgentToolMetadata(permissions=(), effects=ToolEffects(), idempotency=Idempotency.UNKNOWN, data_access=DataAccess.NONE, externality=Externality.LOCAL, timeout=TimeoutPolicy(), result_budget=ResultBudget(), evidence=EvidenceCapture.NONE, approval=ToolApprovalRequirement.DERIVED) dataclass

Typed approval, idempotency, and evidence metadata for a descriptor.

risk property

Return derived risk axes without storing risk as source metadata.

resume property

Return resume metadata derived from stored idempotency.

requires_approval_candidate property

Return whether this tool is a HITL approval candidate.

AgentToolApprovalContext(prompt=None, action_ref=None, metadata=dict()) dataclass

Invocation-specific approval display and correlation metadata.

__post_init__()

Reject blank override strings before they enter approval state.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject blank override strings before they enter approval state."""
    if self.prompt is not None:
        _require_non_blank(self.prompt, "Agent tool approval prompt")
    if self.action_ref is not None:
        _require_non_blank(self.action_ref, "Agent tool approval action ref")

ToolRisk(axes=()) dataclass

Derived typed risk axes for policy and evidence annotations.

requires_approval_candidate property

Return whether the risk is strong enough to suggest HITL approval.

__post_init__()

Reject duplicate axes so risk comparisons stay deterministic.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject duplicate axes so risk comparisons stay deterministic."""
    seen: set[ToolRiskAxis] = set()
    for axis in self.axes:
        if axis in seen:
            raise AgentDefinitionError("Agent tool risk axes must be unique")
        seen.add(axis)

from_metadata(metadata) classmethod

Derive risk axes from source tool metadata.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

@classmethod
def from_metadata(cls, metadata: AgentToolMetadata) -> "ToolRisk":
    """Derive risk axes from source tool metadata."""
    axes: list[ToolRiskAxis] = []
    if metadata.data_access in (DataAccess.READ, DataAccess.READ_WRITE):
        axes.append(ToolRiskAxis.READ)
    if metadata.data_access in (DataAccess.WRITE, DataAccess.READ_WRITE):
        axes.append(ToolRiskAxis.WRITE)
    if metadata.data_access in (DataAccess.WRITE, DataAccess.READ_WRITE):
        axes.append(ToolRiskAxis.SIDE_EFFECT)
    if metadata.externality in (Externality.EXTERNAL, Externality.UNKNOWN):
        axes.append(ToolRiskAxis.SIDE_EFFECT)
    if metadata.effects.destructive:
        axes.append(ToolRiskAxis.DESTRUCTIVE)
    if metadata.effects.network or metadata.externality == Externality.EXTERNAL:
        axes.append(ToolRiskAxis.NETWORK)
    return cls(axes=_deduplicate_axes(axes))

includes(axis)

Return whether this risk contains the requested axis.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def includes(self, axis: ToolRiskAxis) -> bool:
    """Return whether this risk contains the requested axis."""
    return axis in self.axes

AgentToolRuntimeContext(state_id, conversation_id, call_id, tool_name) dataclass

Runtime correlation data injected into framework-owned tool callables.

__post_init__()

Reject runtime context that cannot link events and evidence.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject runtime context that cannot link events and evidence."""
    _require_non_blank(self.state_id, "Agent tool runtime state id")
    _require_non_blank(self.conversation_id, "Agent tool runtime conversation id")
    _require_non_blank(self.call_id, "Agent tool runtime call id")
    _require_non_blank(self.tool_name, "Agent tool runtime tool name")

AgentToolDefinition(name, schema_name, description=None, metadata=AgentToolMetadata()) dataclass

Method-level metadata attached by @agent_tool before owner discovery.

__post_init__()

Reject definitions that would make catalog lookup ambiguous.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject definitions that would make catalog lookup ambiguous."""
    _require_non_blank(self.name, "Agent tool name")
    _require_non_blank(self.schema_name, "Agent tool schema name")
    if self.description is not None and not self.description.strip():
        raise AgentDefinitionError("Agent tool description cannot be blank")

AgentToolDescriptor(identity, owner, callable, schema, description=None, metadata=AgentToolMetadata()) dataclass

Discovered tool descriptor bound to an owner class and callable.

name property

Return the descriptor-local tool name.

bind_invocation(payload)

Bind a decoded model payload before the tool callable is invoked.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def bind_invocation(self, payload: JsonObject) -> "AgentToolBoundInvocation":
    """Bind a decoded model payload before the tool callable is invoked."""
    return bind_agent_tool_invocation(self.callable, payload)

approval_context(payload)

Return invocation-specific approval context for this payload.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def approval_context(self, payload: JsonObject) -> AgentToolApprovalContext:
    """Return invocation-specific approval context for this payload."""
    _ = payload
    return AgentToolApprovalContext()

AgentToolBoundInvocation(args=(), kwargs=dict()) dataclass

Python-call-ready tool invocation arguments.

AgentToolCatalog(descriptors=()) dataclass

Deterministic catalog of descriptors discovered from an Agent class.

__post_init__()

Reject duplicate identity or schema names before model lookup.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def __post_init__(self) -> None:
    """Reject duplicate identity or schema names before model lookup."""
    identity_keys: set[str] = set()
    schema_names: set[str] = set()
    for descriptor in self.descriptors:
        if descriptor.identity.key in identity_keys:
            raise AgentDefinitionError("Agent tool identity must be unique")
        if descriptor.schema.name in schema_names:
            raise AgentDefinitionError("Agent tool schema name must be unique")
        identity_keys.add(descriptor.identity.key)
        schema_names.add(descriptor.schema.name)

by_identity(identity)

Lookup a descriptor by typed identity.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def by_identity(self, identity: AgentToolIdentity) -> AgentToolDescriptor:
    """Lookup a descriptor by typed identity."""
    for descriptor in self.descriptors:
        if descriptor.identity == identity:
            return descriptor
    raise AgentDefinitionError("Agent tool identity is not registered")

by_schema_name(schema_name)

Lookup a descriptor by model-facing schema name.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def by_schema_name(self, schema_name: str) -> AgentToolDescriptor:
    """Lookup a descriptor by model-facing schema name."""
    _require_non_blank(schema_name, "Agent tool schema name")
    for descriptor in self.descriptors:
        if descriptor.schema.name == schema_name:
            return descriptor
    raise AgentDefinitionError("Agent tool schema name is not registered")

agent_tool(*, name=None, schema_name=None, description=None, permissions=(), effects=None, idempotency=Idempotency.UNKNOWN, data_access=None, externality=None, timeout=None, result_budget=None, evidence=EvidenceCapture.NONE, approval=ToolApprovalRequirement.DERIVED)

Attach typed agent-tool metadata to a method object.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def agent_tool(
    *,
    name: str | None = None,
    schema_name: str | None = None,
    description: str | None = None,
    permissions: Sequence[ToolPermission] = (),
    effects: ToolEffects | None = None,
    idempotency: Idempotency = Idempotency.UNKNOWN,
    data_access: DataAccess | None = None,
    externality: Externality | None = None,
    timeout: TimeoutPolicy | None = None,
    result_budget: ResultBudget | None = None,
    evidence: EvidenceCapture = EvidenceCapture.NONE,
    approval: ToolApprovalRequirement = ToolApprovalRequirement.DERIVED,
) -> Callable[[FunctionType], FunctionType]:
    """Attach typed agent-tool metadata to a method object."""

    def decorate(function: FunctionType) -> FunctionType:
        tool_name = _normalize_name(name, function.__name__, "Agent tool name")
        normalized_schema_name = _normalize_name(
            schema_name,
            tool_name,
            "Agent tool schema name",
        )
        tool_effects = effects or ToolEffects()
        metadata = AgentToolMetadata(
            permissions=tuple(permissions),
            effects=tool_effects,
            idempotency=idempotency,
            data_access=data_access or tool_effects.data_access,
            externality=externality or tool_effects.externality,
            timeout=timeout or TimeoutPolicy(),
            result_budget=result_budget or ResultBudget(),
            evidence=evidence,
            approval=approval,
        )
        definition = AgentToolDefinition(
            name=tool_name,
            schema_name=normalized_schema_name,
            description=description,
            metadata=metadata,
        )
        function.__dict__[AGENT_TOOL_DEFINITION_KEY] = definition
        return function

    return decorate

discover_agent_tools(owner)

Discover @agent_tool methods in deterministic class-definition order.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def discover_agent_tools(owner: type[object]) -> AgentToolCatalog:
    """Discover @agent_tool methods in deterministic class-definition order."""
    descriptors: list[AgentToolDescriptor] = []
    resolved_member_names: set[str] = set()
    for declaring_owner in owner.__mro__:
        if declaring_owner is object:
            continue
        for member_name, member in vars(declaring_owner).items():
            if member_name in resolved_member_names:
                continue
            resolved_member_names.add(member_name)
            function = _unwrap_function(member)
            if function is None:
                continue
            definition = get_agent_tool_definition(function)
            if definition is None:
                continue
            descriptors.append(
                _build_descriptor(declaring_owner, function, definition),
            )
    ordered = tuple(
        sorted(
            descriptors,
            key=lambda descriptor: descriptor.identity.key,
        ),
    )
    return AgentToolCatalog(descriptors=ordered)

get_agent_tool_definition(function)

Return decorator metadata attached to a function object.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def get_agent_tool_definition(
    function: FunctionType,
) -> AgentToolDefinition | None:
    """Return decorator metadata attached to a function object."""
    candidate = vars(function).get(AGENT_TOOL_DEFINITION_KEY)
    if candidate is None:
        return None
    if not isinstance(candidate, AgentToolDefinition):
        raise AgentDefinitionError("Agent tool metadata is invalid")
    return candidate

bind_agent_tool_invocation(function, payload)

Bind a structured model payload to a tool function signature.

Source code in core/spakky-agent/src/spakky/agent/tooling.py

def bind_agent_tool_invocation(
    function: AgentToolCallable,
    payload: JsonObject,
) -> AgentToolBoundInvocation:
    """Bind a structured model payload to a tool function signature."""
    invocation = _parse_invocation_payload(payload)
    try:
        function_signature = signature(function)
    except (TypeError, ValueError) as e:
        raise AgentToolBindingError(
            "Agent tool callable signature cannot be inspected"
        ) from e
    call_signature = _drop_injected_parameters(function_signature)
    try:
        bound = call_signature.bind(*invocation.args, **invocation.kwargs)
    except TypeError as e:
        raise AgentToolBindingError(
            "Agent tool invocation payload cannot be bound"
        ) from e
    bound.apply_defaults()
    return AgentToolBoundInvocation(
        args=tuple(bound.args),
        kwargs=dict(bound.kwargs),
    )

Signal Hooks

Declarative signal-hook discovery contracts (ADR-0013 §1).

The framework runner owns the execution loop, but a developer still needs a declarative seam to react to inbound signals (a steering instruction, an external event, a user message) and emit their own stream items without writing a loop body. @on_signal(kind) is that seam: it marks a coroutine generator method the same way @agent_tool marks a tool method, and the runner invokes every matching hook when it consumes a signal of that kind at a poll point.

Discovery mirrors discover_agent_tools exactly — an MRO walk reading vars() (not the banned getattr) so the raw function objects are found before descriptor binding — so the two declarative seams stay structurally identical for a reader.

AgentSignalHookDefinition(kind) dataclass

Method-level metadata attached by @on_signal before owner discovery.

AgentSignalHookIdentity(owner_module, owner_qualname, member_name) dataclass

Hook identity independent from the developer-chosen method name.

key property

Return a stable key for deterministic ordering and logs.

AgentSignalHookDescriptor(identity, owner, callable, kind) dataclass

Discovered signal hook bound to an owner class and callable.

AgentSignalHookCatalog(descriptors=()) dataclass

Deterministic catalog of signal hooks discovered from an Agent class.

__post_init__()

Reject duplicate hook identities so dispatch order stays stable.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def __post_init__(self) -> None:
    """Reject duplicate hook identities so dispatch order stays stable."""
    identity_keys: set[str] = set()
    for descriptor in self.descriptors:
        if descriptor.identity.key in identity_keys:
            raise AgentDefinitionError("Agent signal hook identity must be unique")
        identity_keys.add(descriptor.identity.key)

hooks_for(kind)

Return every hook that handles the requested signal kind, in order.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def hooks_for(
    self,
    kind: AgentSignalKind,
) -> tuple[AgentSignalHookDescriptor, ...]:
    """Return every hook that handles the requested signal kind, in order."""
    return tuple(
        descriptor for descriptor in self.descriptors if descriptor.kind is kind
    )

on_signal(kind)

Declare a method that reacts to one inbound signal kind.

The decorated method must be an async generator yielding AgentYield items and accepting exactly one signal: AgentSignal argument besides self. The runner invokes it when it consumes a signal of kind at a poll point, and forwards every yielded item into the public stream.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def on_signal(kind: AgentSignalKind) -> Callable[[FunctionType], FunctionType]:
    """Declare a method that reacts to one inbound signal kind.

    The decorated method must be an async generator yielding ``AgentYield``
    items and accepting exactly one ``signal: AgentSignal`` argument besides
    ``self``. The runner invokes it when it consumes a signal of ``kind`` at a
    poll point, and forwards every yielded item into the public stream.
    """

    def decorate(function: FunctionType) -> FunctionType:
        function.__dict__[AGENT_SIGNAL_HOOK_DEFINITION_KEY] = AgentSignalHookDefinition(
            kind=kind,
        )
        return function

    return decorate

discover_agent_signal_hooks(owner)

Discover @on_signal methods in deterministic class-definition order.

Source code in core/spakky-agent/src/spakky/agent/hooks.py

def discover_agent_signal_hooks(owner: type[object]) -> AgentSignalHookCatalog:
    """Discover ``@on_signal`` methods in deterministic class-definition order."""
    descriptors: list[AgentSignalHookDescriptor] = []
    resolved_member_names: set[str] = set()
    for declaring_owner in owner.__mro__:
        if declaring_owner is object:
            continue
        for member_name, member in vars(declaring_owner).items():
            if member_name in resolved_member_names:
                continue
            resolved_member_names.add(member_name)
            if not isinstance(member, FunctionType):
                continue
            definition = _get_signal_hook_definition(member)
            if definition is None:
                continue
            _validate_hook_contract(member)
            descriptors.append(
                AgentSignalHookDescriptor(
                    identity=AgentSignalHookIdentity(
                        owner_module=declaring_owner.__module__,
                        owner_qualname=declaring_owner.__qualname__,
                        member_name=member_name,
                    ),
                    owner=declaring_owner,
                    callable=member,
                    kind=definition.kind,
                ),
            )
    ordered = tuple(
        sorted(descriptors, key=lambda descriptor: descriptor.identity.key),
    )
    return AgentSignalHookCatalog(descriptors=ordered)

Yield

Agent streaming yield contracts.

TextDelta = Token module-attribute

Backward-compatible token-yield payload alias.

Message = Progress module-attribute

Backward-compatible progress-yield payload alias.

AgentYieldKind

Bases: StrEnum

Canonical public vocabulary yielded by agent execution.

AgentYield(kind, payload) dataclass

Typed stream item returned from an agent execute generator.

Token(text, metadata=dict()) dataclass

Incremental model token intended for streaming clients.

Progress(message, current_step=None, metadata=dict()) dataclass

Agent progress update intended for direct inbound adapter consumption.

Tool(name, call_id=None, arguments=dict(), result=None, metadata=dict()) dataclass

Tool call or tool result surfaced by agent execution.

Evidence(evidence, metadata=dict()) dataclass

Evidence item surfaced to the inbound adapter.

Approval(id, prompt, allowed_decisions, metadata) dataclass

Approval request surfaced to the inbound adapter.

Error(code, message, retryable=False, metadata=dict()) dataclass

Recoverable or terminal execution error surfaced to the caller.

Cancel(reason=None, requested_by=None, metadata=dict()) dataclass

Cancellation acknowledgement surfaced to the caller.

Final(output, metadata) dataclass

Final output carried by a generator stream.

Model Interface

Agent public interface ports.

ContextDigest(id, context_identity, source_manifest_ref, digest, derived_from_pack_ids=tuple(), compression_evidence_ref=None, algorithm=None, summary=None, created_at=None, metadata=dict()) dataclass

Derived compression evidence for a context identity.

ContextFreshness

Bases: StrEnum

Freshness classification for context rot and budget decisions.

ContextManifest(id, entries, origin_ref=None, evidence_refs=tuple(), created_at=None, metadata=dict()) dataclass

Auditable composition record for model input context packs.

ContextManifestEntry(pack_id, source, role, origin_ref, evidence_ref=None, digest_ref=None, sensitive_fields=(), metadata=dict()) dataclass

One audited pack entry inside a context manifest.

ContextPack(id, content, source, role, freshness=ContextFreshness.UNKNOWN, relevance=None, token_budget=ContextTokenBudget(), sensitivity=ContextSensitivity.INTERNAL, sensitive_fields=(), metadata=dict()) dataclass

LLM-facing context unit derived from state, signal, or evidence.

guarded_content(policy=None)

Return deterministic model-safe content for this context pack.

Source code in core/spakky-agent/src/spakky/agent/context.py

def guarded_content(
    self,
    policy: ContextExposurePolicy | None = None,
) -> str:
    """Return deterministic model-safe content for this context pack."""
    exposure_policy = policy or ContextExposurePolicy()
    if self.sensitivity == ContextSensitivity.REDACTED:
        return "[REDACTED]"
    guarded = guard_json_value(
        {"content": self.content},
        tuple(
            SensitiveFieldDescriptor(
                ("content", *descriptor.path), descriptor.field
            )
            for descriptor in self.sensitive_fields
        ),
        exposure_policy,
    )
    content = cast(Mapping[str, JsonValue], guarded).get("content")
    if isinstance(content, str):
        return content
    return "[REDACTED]"

message_metadata(policy=None)

Return non-content metadata for provider-neutral model messages.

Source code in core/spakky-agent/src/spakky/agent/context.py

def message_metadata(
    self,
    policy: ContextExposurePolicy | None = None,
) -> Mapping[str, JsonValue]:
    """Return non-content metadata for provider-neutral model messages."""
    exposure_policy = policy or ContextExposurePolicy()
    metadata: dict[str, JsonValue] = {
        "context_pack_id": self.id,
        "source": self.source,
        "role": self.role.value,
        "freshness": self.freshness.value,
        "relevance": self.relevance,
        "token_budget": {
            "max_tokens": self.token_budget.max_tokens,
            "estimated_tokens": self.token_budget.estimated_tokens,
            "reserved_output_tokens": self.token_budget.reserved_output_tokens,
        },
        "sensitivity": self.sensitivity.value,
        "metadata": self.metadata,
    }
    if exposure_policy.include_sensitive_context_metadata and self.sensitive_fields:
        metadata["sensitive_fields"] = tuple(
            descriptor.to_metadata() for descriptor in self.sensitive_fields
        )
    return metadata

ContextPackRole

Bases: StrEnum

Semantic role of a context pack inside a model request.

ContextSensitivity

Bases: StrEnum

Deterministic sensitivity metadata carried before model input.

ContextTokenBudget(max_tokens=None, estimated_tokens=None, reserved_output_tokens=None) dataclass

Token budget allocated to one context pack.

IAgentModel

Bases: ABC

Outbound model adapter port owned by spakky-agent core.

capability abstractmethod property

Return the backend capability descriptor queryable before a run.

capability_for(selection=None)

Return capability for a run-specific model selection.

Existing fixed-model adapters can ignore the selector and inherit the default. Routing adapters can override this to expose per-model context windows, reasoning support, or token-counting support before a request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def capability_for(
    self,
    selection: ModelSelection | None = None,
) -> ModelCapability:
    """Return capability for a run-specific model selection.

    Existing fixed-model adapters can ignore the selector and inherit the
    default. Routing adapters can override this to expose per-model context
    windows, reasoning support, or token-counting support before a request.
    """
    _ = selection
    return self.capability

complete(request) abstractmethod async

Return a complete model response for the request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

@abstractmethod
async def complete(self, request: ModelRequest) -> ModelResponse:
    """Return a complete model response for the request."""
    ...

stream(request) abstractmethod

Return provider-neutral stream events for the request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

@abstractmethod
def stream(self, request: ModelRequest) -> AsyncIterator[ModelStreamEvent]:
    """Return provider-neutral stream events for the request."""
    ...

JsonSchemaConstraint(schema, strict=True) dataclass

JSON schema constraint shared by structured output and tool calling.

ModelCapability(supports_reasoning=False, context_window_tokens=None, supports_token_counting=False) dataclass

Provider-neutral declaration of a model backend's queryable abilities.

The agent runner consults this descriptor before a run to adjust behaviour without invoking the backend. supports_reasoning gates whether the runner expects REASONING_DELTA events; when False the adapter omits them rather than failing (graceful degrade). context_window_tokens is None when the backend does not declare a fixed limit. supports_token_counting declares whether the backend can report token accounting for a request before sending it.

ModelError(code, message, retryable=False, metadata=dict()) dataclass

Provider-neutral model failure payload.

ModelMessage(role, content, metadata=dict()) dataclass

Provider-neutral model message.

ModelMessageRole

Bases: StrEnum

Roles accepted by provider-neutral model messages.

ModelRequest(messages, context=tuple(), context_manifest=None, context_digest=None, structured_output=None, tool_calling=None, sampling=SamplingOptions(), streaming=StreamingOptions(), model_selection=None, metadata=dict()) dataclass

Provider-neutral request passed to an agent model adapter.

assemble_messages(policy=None)

Assemble prompt messages from typed context packs without concatenation.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def assemble_messages(
    self,
    policy: ContextExposurePolicy | None = None,
) -> tuple[ModelMessage, ...]:
    """Assemble prompt messages from typed context packs without concatenation."""
    exposure_policy = policy or ContextExposurePolicy()
    context_messages = tuple(
        ModelMessage(
            role=ModelMessageRole.EVIDENCE,
            content=pack.guarded_content(exposure_policy),
            metadata=pack.message_metadata(exposure_policy),
        )
        for pack in self.context
    )
    return (*self.messages, *context_messages)

ModelResponse(content, structured_output=None, tool_calls=tuple(), usage=ModelUsage(), metadata=dict()) dataclass

Provider-neutral non-streaming model response.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive output payloads deterministically guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelResponse":
    """Return a copy with sensitive output payloads deterministically guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    content = self.content
    structured_output = self.structured_output
    content_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path in ((), ("content",))
    )
    if content_descriptors:
        guarded_content = guard_json_value(
            {"content": content},
            tuple(
                SensitiveFieldDescriptor(("content",), descriptor.field)
                for descriptor in content_descriptors
            ),
            exposure_policy,
        )
        content_value = cast(Mapping[str, JsonValue], guarded_content).get(
            "content"
        )
        if isinstance(content_value, str):
            content = content_value
        else:
            content = "[REDACTED]"
    structured_output_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path not in ((), ("content",))
    )
    if structured_output_descriptors:
        structured_output = guard_json_value(
            structured_output,
            structured_output_descriptors,
            exposure_policy,
        )
    return replace(self, content=content, structured_output=structured_output)

ModelStreamEvent(kind, token_delta=None, message_delta=None, reasoning_delta=None, tool_call=None, tool_call_args_delta=None, structured_output=None, error=None, usage=None, metadata=dict()) dataclass

Provider-neutral model streaming event.

token_delta carries the generic streamed token channel. message_delta and reasoning_delta distinguish assistant-facing text from model reasoning so callers can route or suppress reasoning independently. tool_call_args_delta carries incremental tool-call argument text framed by TOOL_CALL_START and TOOL_CALL_END boundary events that reference the same tool_call.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive streaming payloads guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelStreamEvent":
    """Return a copy with sensitive streaming payloads guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    text_channels: dict[str, str | None] = {
        "token_delta": self.token_delta,
        "message_delta": self.message_delta,
        "reasoning_delta": self.reasoning_delta,
        "tool_call_args_delta": self.tool_call_args_delta,
    }
    guarded_text = {
        field_name: self._guard_text_delta(
            field_name, value, sensitive_fields, exposure_policy
        )
        for field_name, value in text_channels.items()
        if value is not None
    }
    text_paths = (
        (),
        *((field_name,) for field_name in text_channels),
    )
    structured_output = self.structured_output
    structured_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path not in text_paths
    )
    if structured_descriptors:
        structured_output = guard_json_value(
            structured_output,
            structured_descriptors,
            exposure_policy,
        )
    tool_call = self.tool_call
    if tool_call is not None:
        tool_call = tool_call.guarded(sensitive_fields, exposure_policy)
    return replace(
        self,
        structured_output=structured_output,
        tool_call=tool_call,
        **guarded_text,
    )

ModelStreamEventKind

Bases: StrEnum

Provider-neutral streaming event kinds emitted by a model adapter.

ModelToolCall(name, arguments, call_id=None, metadata=dict()) dataclass

Tool invocation candidate emitted by a model adapter.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive argument values deterministically guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelToolCall":
    """Return a copy with sensitive argument values deterministically guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    guarded_arguments = guard_json_value(
        self.arguments,
        sensitive_fields,
        exposure_policy,
    )
    if not isinstance(guarded_arguments, Mapping):
        guarded_arguments = {}
    return replace(self, arguments=guarded_arguments)

ModelToolChoice

Bases: StrEnum

Provider-neutral tool calling strategy requested from a model adapter.

ModelToolSpec(name, parameters, description=None, metadata=dict()) dataclass

LLM-facing tool descriptor normalized by agent tooling.

ModelUsage(input_tokens=None, output_tokens=None, total_tokens=None) dataclass

Token accounting reported by a model adapter.

SamplingOptions(temperature=None, top_p=None, max_tokens=None) dataclass

Portable model sampling options.

StreamingOptions(include_usage=True, include_progress=True) dataclass

Portable model streaming options.

StructuredOutputSpec(constraint, output_type_name=None) dataclass

Structured output contract requested from a model adapter.

ToolCallingSpec(tools, choice=ModelToolChoice.AUTO) dataclass

Tool calling contract requested from a model adapter.

IAgentEvidenceRepository

Bases: ABC

Append-only evidence repository exposed to agent-facing code.

append(evidence) abstractmethod

Append immutable evidence and return the stored artifact.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def append(self, evidence: AgentEvidence) -> AgentEvidence:
    """Append immutable evidence and return the stored artifact."""
    ...

get(evidence_id) abstractmethod

Return one evidence artifact by id.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get(self, evidence_id: str) -> AgentEvidence:
    """Return one evidence artifact by id."""
    ...

list_by_state(state_id) abstractmethod

Return evidence artifacts captured for an agent state.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_state(self, state_id: str) -> Sequence[AgentEvidence]:
    """Return evidence artifacts captured for an agent state."""
    ...

list_by_manifest_ref(manifest_ref) abstractmethod

Return evidence artifacts associated with a context manifest.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_manifest_ref(self, manifest_ref: str) -> Sequence[AgentEvidence]:
    """Return evidence artifacts associated with a context manifest."""
    ...

IAgentSignalRepository

Bases: ABC

Durable inbound queue repository for agent signals.

append(signal) abstractmethod

Append a signal to the inbound queue and return the stored entry.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def append(self, signal: AgentSignal) -> AgentSignal:
    """Append a signal to the inbound queue and return the stored entry."""
    ...

list_pending(state_id) abstractmethod

Return unconsumed signals for an agent state in append/queue order.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_pending(self, state_id: str) -> Sequence[AgentSignal]:
    """Return unconsumed signals for an agent state in append/queue order."""
    ...

mark_consumed(signal_id) abstractmethod

Mark one queued signal as consumed after a non-blocking poll.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def mark_consumed(self, signal_id: str) -> AgentSignal:
    """Mark one queued signal as consumed after a non-blocking poll."""
    ...

IAgentStateRepository

Bases: ABC

Materialized state repository for long-running agent executions.

get(state_id) abstractmethod

Return a persisted agent state by id.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get(self, state_id: str) -> AgentState:
    """Return a persisted agent state by id."""
    ...

get_or_none(state_id) abstractmethod

Return a persisted agent state by id, or None when absent.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get_or_none(self, state_id: str) -> AgentState | None:
    """Return a persisted agent state by id, or None when absent."""
    ...

save(state) abstractmethod

Persist the materialized state and return the saved snapshot.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def save(self, state: AgentState) -> AgentState:
    """Persist the materialized state and return the saved snapshot."""
    ...

list_by_status(status) abstractmethod

Return states matching an externally observable lifecycle status.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_status(self, status: AgentStatus) -> Sequence[AgentState]:
    """Return states matching an externally observable lifecycle status."""
    ...

list_resume_candidates() abstractmethod

Return active or interrupted states that may resume after restart.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_resume_candidates(self) -> Sequence[AgentState]:
    """Return active or interrupted states that may resume after restart."""
    ...

ConversationTurn(role, content, metadata=dict()) dataclass

One persisted exchange in a multi-turn conversation transcript.

A turn is the protocol-neutral unit a TaskStore persists: who spoke (role) and what was said (content). It is intentionally narrower than a model-request ModelMessage — the transcript records the user/assistant dialogue that seeds future turns, not the system or evidence framing the runner assembles fresh on each request. as_model_message projects a turn back into the model-request vocabulary when the runner replays history.

__post_init__()

Reject turns that cannot seed a future model request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

def __post_init__(self) -> None:
    """Reject turns that cannot seed a future model request."""
    if self.role not in (ModelMessageRole.USER, ModelMessageRole.ASSISTANT):
        raise AgentDefinitionError(
            "Conversation turn role must be user or assistant"
        )
    if not self.content.strip():
        raise AgentDefinitionError("Conversation turn content cannot be blank")

as_model_message()

Project this turn into a model-request message for history replay.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

def as_model_message(self) -> ModelMessage:
    """Project this turn into a model-request message for history replay."""
    return ModelMessage(
        role=self.role, content=self.content, metadata=self.metadata
    )

ITaskStore

Bases: ABC

Durable conversation-history store keyed by conversation_id.

Persists the running transcript of a server-side session so a later run with the same conversation_id continues the conversation (ADR-0013 §6). A2A can map its contextId to this key for transcript replay, while A2A task snapshots are stored by spakky-a2a's repository bridge.

load_history(conversation_id) abstractmethod

Return the persisted transcript for a conversation in turn order.

Returns an empty sequence for a conversation that has no persisted turns yet — the first turn of a brand-new conversation.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

@abstractmethod
def load_history(self, conversation_id: str) -> Sequence[ConversationTurn]:
    """Return the persisted transcript for a conversation in turn order.

    Returns an empty sequence for a conversation that has no persisted turns
    yet — the first turn of a brand-new conversation.
    """
    ...

append_turns(conversation_id, turns) abstractmethod

Append new turns to a conversation's transcript in order.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

@abstractmethod
def append_turns(
    self,
    conversation_id: str,
    turns: Sequence[ConversationTurn],
) -> None:
    """Append new turns to a conversation's transcript in order."""
    ...

Provider-neutral agent model port.

ModelMessageRole

Bases: StrEnum

Roles accepted by provider-neutral model messages.

ModelMessage(role, content, metadata=dict()) dataclass

Provider-neutral model message.

JsonSchemaConstraint(schema, strict=True) dataclass

JSON schema constraint shared by structured output and tool calling.

StructuredOutputSpec(constraint, output_type_name=None) dataclass

Structured output contract requested from a model adapter.

ModelToolChoice

Bases: StrEnum

Provider-neutral tool calling strategy requested from a model adapter.

ModelToolSpec(name, parameters, description=None, metadata=dict()) dataclass

LLM-facing tool descriptor normalized by agent tooling.

ToolCallingSpec(tools, choice=ModelToolChoice.AUTO) dataclass

Tool calling contract requested from a model adapter.

SamplingOptions(temperature=None, top_p=None, max_tokens=None) dataclass

Portable model sampling options.

StreamingOptions(include_usage=True, include_progress=True) dataclass

Portable model streaming options.

ModelSelection(provider=None, model=None, profile=None, metadata=dict()) dataclass

Provider/model selector carried by one Agent run.

A service may let a user choose OpenAI, Anthropic, Vertex, OpenRouter, vLLM, or another provider per run. The selector is intentionally provider-neutral: concrete adapters or routing models decide which values they accept.

__post_init__()

Reject blank selector fields before they reach provider adapters.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def __post_init__(self) -> None:
    """Reject blank selector fields before they reach provider adapters."""
    for label, value in (
        ("provider", self.provider),
        ("model", self.model),
        ("profile", self.profile),
    ):
        if value is not None and not value.strip():
            raise AgentDefinitionError(f"Model selection {label} cannot be blank")

ModelRequest(messages, context=tuple(), context_manifest=None, context_digest=None, structured_output=None, tool_calling=None, sampling=SamplingOptions(), streaming=StreamingOptions(), model_selection=None, metadata=dict()) dataclass

Provider-neutral request passed to an agent model adapter.

assemble_messages(policy=None)

Assemble prompt messages from typed context packs without concatenation.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def assemble_messages(
    self,
    policy: ContextExposurePolicy | None = None,
) -> tuple[ModelMessage, ...]:
    """Assemble prompt messages from typed context packs without concatenation."""
    exposure_policy = policy or ContextExposurePolicy()
    context_messages = tuple(
        ModelMessage(
            role=ModelMessageRole.EVIDENCE,
            content=pack.guarded_content(exposure_policy),
            metadata=pack.message_metadata(exposure_policy),
        )
        for pack in self.context
    )
    return (*self.messages, *context_messages)

ModelUsage(input_tokens=None, output_tokens=None, total_tokens=None) dataclass

Token accounting reported by a model adapter.

ModelToolCall(name, arguments, call_id=None, metadata=dict()) dataclass

Tool invocation candidate emitted by a model adapter.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive argument values deterministically guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelToolCall":
    """Return a copy with sensitive argument values deterministically guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    guarded_arguments = guard_json_value(
        self.arguments,
        sensitive_fields,
        exposure_policy,
    )
    if not isinstance(guarded_arguments, Mapping):
        guarded_arguments = {}
    return replace(self, arguments=guarded_arguments)

ModelError(code, message, retryable=False, metadata=dict()) dataclass

Provider-neutral model failure payload.

ModelResponse(content, structured_output=None, tool_calls=tuple(), usage=ModelUsage(), metadata=dict()) dataclass

Provider-neutral non-streaming model response.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive output payloads deterministically guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelResponse":
    """Return a copy with sensitive output payloads deterministically guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    content = self.content
    structured_output = self.structured_output
    content_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path in ((), ("content",))
    )
    if content_descriptors:
        guarded_content = guard_json_value(
            {"content": content},
            tuple(
                SensitiveFieldDescriptor(("content",), descriptor.field)
                for descriptor in content_descriptors
            ),
            exposure_policy,
        )
        content_value = cast(Mapping[str, JsonValue], guarded_content).get(
            "content"
        )
        if isinstance(content_value, str):
            content = content_value
        else:
            content = "[REDACTED]"
    structured_output_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path not in ((), ("content",))
    )
    if structured_output_descriptors:
        structured_output = guard_json_value(
            structured_output,
            structured_output_descriptors,
            exposure_policy,
        )
    return replace(self, content=content, structured_output=structured_output)

ModelStreamEventKind

Bases: StrEnum

Provider-neutral streaming event kinds emitted by a model adapter.

ModelStreamEvent(kind, token_delta=None, message_delta=None, reasoning_delta=None, tool_call=None, tool_call_args_delta=None, structured_output=None, error=None, usage=None, metadata=dict()) dataclass

Provider-neutral model streaming event.

token_delta carries the generic streamed token channel. message_delta and reasoning_delta distinguish assistant-facing text from model reasoning so callers can route or suppress reasoning independently. tool_call_args_delta carries incremental tool-call argument text framed by TOOL_CALL_START and TOOL_CALL_END boundary events that reference the same tool_call.

guarded(sensitive_fields, policy=None)

Return a copy with sensitive streaming payloads guarded.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def guarded(
    self,
    sensitive_fields: Sequence[SensitiveFieldDescriptor],
    policy: EvidenceExposurePolicy | None = None,
) -> "ModelStreamEvent":
    """Return a copy with sensitive streaming payloads guarded."""
    exposure_policy = policy or EvidenceExposurePolicy()
    text_channels: dict[str, str | None] = {
        "token_delta": self.token_delta,
        "message_delta": self.message_delta,
        "reasoning_delta": self.reasoning_delta,
        "tool_call_args_delta": self.tool_call_args_delta,
    }
    guarded_text = {
        field_name: self._guard_text_delta(
            field_name, value, sensitive_fields, exposure_policy
        )
        for field_name, value in text_channels.items()
        if value is not None
    }
    text_paths = (
        (),
        *((field_name,) for field_name in text_channels),
    )
    structured_output = self.structured_output
    structured_descriptors = tuple(
        descriptor
        for descriptor in sensitive_fields
        if descriptor.path not in text_paths
    )
    if structured_descriptors:
        structured_output = guard_json_value(
            structured_output,
            structured_descriptors,
            exposure_policy,
        )
    tool_call = self.tool_call
    if tool_call is not None:
        tool_call = tool_call.guarded(sensitive_fields, exposure_policy)
    return replace(
        self,
        structured_output=structured_output,
        tool_call=tool_call,
        **guarded_text,
    )

ModelCapability(supports_reasoning=False, context_window_tokens=None, supports_token_counting=False) dataclass

Provider-neutral declaration of a model backend's queryable abilities.

The agent runner consults this descriptor before a run to adjust behaviour without invoking the backend. supports_reasoning gates whether the runner expects REASONING_DELTA events; when False the adapter omits them rather than failing (graceful degrade). context_window_tokens is None when the backend does not declare a fixed limit. supports_token_counting declares whether the backend can report token accounting for a request before sending it.

IAgentModel

Bases: ABC

Outbound model adapter port owned by spakky-agent core.

capability abstractmethod property

Return the backend capability descriptor queryable before a run.

capability_for(selection=None)

Return capability for a run-specific model selection.

Existing fixed-model adapters can ignore the selector and inherit the default. Routing adapters can override this to expose per-model context windows, reasoning support, or token-counting support before a request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

def capability_for(
    self,
    selection: ModelSelection | None = None,
) -> ModelCapability:
    """Return capability for a run-specific model selection.

    Existing fixed-model adapters can ignore the selector and inherit the
    default. Routing adapters can override this to expose per-model context
    windows, reasoning support, or token-counting support before a request.
    """
    _ = selection
    return self.capability

complete(request) abstractmethod async

Return a complete model response for the request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

@abstractmethod
async def complete(self, request: ModelRequest) -> ModelResponse:
    """Return a complete model response for the request."""
    ...

stream(request) abstractmethod

Return provider-neutral stream events for the request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/model.py

@abstractmethod
def stream(self, request: ModelRequest) -> AsyncIterator[ModelStreamEvent]:
    """Return provider-neutral stream events for the request."""
    ...

Persistence ports for durable agent state, signal, and evidence.

IAgentStateRepository

Bases: ABC

Materialized state repository for long-running agent executions.

get(state_id) abstractmethod

Return a persisted agent state by id.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get(self, state_id: str) -> AgentState:
    """Return a persisted agent state by id."""
    ...

get_or_none(state_id) abstractmethod

Return a persisted agent state by id, or None when absent.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get_or_none(self, state_id: str) -> AgentState | None:
    """Return a persisted agent state by id, or None when absent."""
    ...

save(state) abstractmethod

Persist the materialized state and return the saved snapshot.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def save(self, state: AgentState) -> AgentState:
    """Persist the materialized state and return the saved snapshot."""
    ...

list_by_status(status) abstractmethod

Return states matching an externally observable lifecycle status.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_status(self, status: AgentStatus) -> Sequence[AgentState]:
    """Return states matching an externally observable lifecycle status."""
    ...

list_resume_candidates() abstractmethod

Return active or interrupted states that may resume after restart.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_resume_candidates(self) -> Sequence[AgentState]:
    """Return active or interrupted states that may resume after restart."""
    ...

IAgentSignalRepository

Bases: ABC

Durable inbound queue repository for agent signals.

append(signal) abstractmethod

Append a signal to the inbound queue and return the stored entry.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def append(self, signal: AgentSignal) -> AgentSignal:
    """Append a signal to the inbound queue and return the stored entry."""
    ...

list_pending(state_id) abstractmethod

Return unconsumed signals for an agent state in append/queue order.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_pending(self, state_id: str) -> Sequence[AgentSignal]:
    """Return unconsumed signals for an agent state in append/queue order."""
    ...

mark_consumed(signal_id) abstractmethod

Mark one queued signal as consumed after a non-blocking poll.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def mark_consumed(self, signal_id: str) -> AgentSignal:
    """Mark one queued signal as consumed after a non-blocking poll."""
    ...

IAgentEvidenceRepository

Bases: ABC

Append-only evidence repository exposed to agent-facing code.

append(evidence) abstractmethod

Append immutable evidence and return the stored artifact.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def append(self, evidence: AgentEvidence) -> AgentEvidence:
    """Append immutable evidence and return the stored artifact."""
    ...

get(evidence_id) abstractmethod

Return one evidence artifact by id.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def get(self, evidence_id: str) -> AgentEvidence:
    """Return one evidence artifact by id."""
    ...

list_by_state(state_id) abstractmethod

Return evidence artifacts captured for an agent state.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_state(self, state_id: str) -> Sequence[AgentEvidence]:
    """Return evidence artifacts captured for an agent state."""
    ...

list_by_manifest_ref(manifest_ref) abstractmethod

Return evidence artifacts associated with a context manifest.

Source code in core/spakky-agent/src/spakky/agent/interfaces/repository.py

@abstractmethod
def list_by_manifest_ref(self, manifest_ref: str) -> Sequence[AgentEvidence]:
    """Return evidence artifacts associated with a context manifest."""
    ...

Persistence port for multi-turn conversation history (ADR-0013 §6).

ADR-0013 §6 supports multi-turn conversations through two paths that the framework runner reconciles:

• server-side persisted sessions — the framework persists the running transcript and continues it by conversation_id. This port is that conversation-history persistence contract. It is protocol-neutral, but it is not the A2A task snapshot repository: spakky-a2a owns A2A Task storage through its plugin repository and bridges it to a2a-sdk TaskStore.
• client-injected history — a stateless caller passes the prior transcript on each run (RunAgentInput.message_history) and no store is consulted.

The conversation_id key is the protocol-neutral thread identifier carried by every event (AgentEventAttribution.conversation_id) and seeded by RunAgentInput.effective_conversation_id. AG-UI projects it as threadId and A2A projects it as contextId when an adapter wants transcript replay; A2A protocol Task records remain plugin-owned snapshots.

ConversationTurn(role, content, metadata=dict()) dataclass

One persisted exchange in a multi-turn conversation transcript.

A turn is the protocol-neutral unit a TaskStore persists: who spoke (role) and what was said (content). It is intentionally narrower than a model-request ModelMessage — the transcript records the user/assistant dialogue that seeds future turns, not the system or evidence framing the runner assembles fresh on each request. as_model_message projects a turn back into the model-request vocabulary when the runner replays history.

__post_init__()

Reject turns that cannot seed a future model request.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

def __post_init__(self) -> None:
    """Reject turns that cannot seed a future model request."""
    if self.role not in (ModelMessageRole.USER, ModelMessageRole.ASSISTANT):
        raise AgentDefinitionError(
            "Conversation turn role must be user or assistant"
        )
    if not self.content.strip():
        raise AgentDefinitionError("Conversation turn content cannot be blank")

as_model_message()

Project this turn into a model-request message for history replay.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

def as_model_message(self) -> ModelMessage:
    """Project this turn into a model-request message for history replay."""
    return ModelMessage(
        role=self.role, content=self.content, metadata=self.metadata
    )

ITaskStore

Bases: ABC

Durable conversation-history store keyed by conversation_id.

Persists the running transcript of a server-side session so a later run with the same conversation_id continues the conversation (ADR-0013 §6). A2A can map its contextId to this key for transcript replay, while A2A task snapshots are stored by spakky-a2a's repository bridge.

load_history(conversation_id) abstractmethod

Return the persisted transcript for a conversation in turn order.

Returns an empty sequence for a conversation that has no persisted turns yet — the first turn of a brand-new conversation.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

@abstractmethod
def load_history(self, conversation_id: str) -> Sequence[ConversationTurn]:
    """Return the persisted transcript for a conversation in turn order.

    Returns an empty sequence for a conversation that has no persisted turns
    yet — the first turn of a brand-new conversation.
    """
    ...

append_turns(conversation_id, turns) abstractmethod

Append new turns to a conversation's transcript in order.

Source code in core/spakky-agent/src/spakky/agent/interfaces/task_store.py

@abstractmethod
def append_turns(
    self,
    conversation_id: str,
    turns: Sequence[ConversationTurn],
) -> None:
    """Append new turns to a conversation's transcript in order."""
    ...

Types

Shared JSON-compatible type aliases for agent contracts.

JsonPrimitive = bool | float | int | str | None

Scalar JSON value accepted by agent public contracts.

JsonValue = JsonPrimitive | Mapping[str, JsonValue] | Sequence[JsonValue]

Recursive JSON-compatible value used at model, signal, and evidence boundaries.

JsonObject = Mapping[str, JsonValue]

JSON object payload used by public agent contracts.

Plugin

Plugin initialization entry point for spakky-agent.

initialize(app)

Initialize spakky-agent core contracts.

The package intentionally registers no persistence implementation; production repositories must arrive through feature contributions.

Source code in core/spakky-agent/src/spakky/agent/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize spakky-agent core contracts.

    The package intentionally registers no persistence implementation; production
    repositories must arrive through feature contributions.
    """
    app.add(AgentRunnerFactory)
    app.add(AgentBootstrapValidationPostProcessor)

Bootstrap validation for Agent Pod instances.

AgentBootstrapValidationPostProcessor()

Bases: IPostProcessor, IContainerAware

Validate Agent metadata during application bootstrap.

Initialize without assuming application context injection happened.

Source code in core/spakky-agent/src/spakky/agent/post_processor.py

def __init__(self) -> None:
    """Initialize without assuming application context injection happened."""
    self._container = None

set_container(container)

Receive the active container for contribution validation.

Source code in core/spakky-agent/src/spakky/agent/post_processor.py

@override
def set_container(self, container: IContainer) -> None:
    """Receive the active container for contribution validation."""
    self._container = container

post_process(pod)

Fail startup when an Agent contract is no longer valid.

Source code in core/spakky-agent/src/spakky/agent/post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Fail startup when an Agent contract is no longer valid."""
    agent = Agent.get_or_none(type(pod))
    if agent is None:
        return pod
    try:
        agent.validate_bootstrap()
    except AgentDefinitionError as e:
        raise AgentBootstrapError("Agent bootstrap contract is invalid") from e
    self._validate_persistence_contributions(agent)
    return pod

에러

Error classes for the spakky-agent package.

AbstractSpakkyAgentError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for agent-related errors.

AgentDefinitionError

Bases: AbstractSpakkyAgentError

Raised when an agent contract cannot be defined safely.

AgentToolBindingError

Bases: AbstractSpakkyAgentError

Raised when a model tool-call payload cannot be bound safely.

AgentToolDispatchError

Bases: AbstractSpakkyAgentError

Raised when a model tool call names a tool absent from the catalog.

AgentBootstrapError

Bases: AbstractSpakkyAgentError

Raised when agent bootstrap validation fails.

AgentPersistenceConfigurationError

Bases: AgentBootstrapError

Raised when production agent persistence is required but not provided.

AgentModelConfigurationError

Bases: AgentBootstrapError

Raised when an agent requires a model adapter but none is registered.

AgentOutputGuardError

Bases: AbstractSpakkyAgentError

Raised when an output guard detects unsafe streaming exposure.