Skip to main content

Runtime

Protolink's runtime primitives provide a stable execution layer above the core A2A-derived Task, Message, Part, and Artifact models. They are intentionally generic: the same contracts work for local CLIs, workflow engines, support assistants, research systems, browser agents, data tools, and any other agent application.

The runtime layer does not replace transports, telemetry, storage, or structured flows. It gives them shared execution metadata, concrete action intents, policy and approval boundaries, and a normalized event stream.

Runtime control layerRuntime Primitives

The application-facing contracts for run context, cancellation, budgets, policies, approvals, actions, normalized events, reports, replay, regression comparison, and redaction.

protolink.runtime
RunContextRunBudgetPolicyDecisionApprovalRequestRunEventRunReportRunReportDiffConfig
ContextAttach run, session, trace, workspace, permission, and budget metadata to tasks without ad hoc keys.RunContext
ControlCancel active work and enforce model, tool, token, step, and runtime limits before side effects occur.CancellationToken
PolicyEvaluate actions as allow, deny, or require approval with preview artifacts before execution.CapabilityPolicy
Reports and regressionReplay recorded facts safely and compare normalized reports without repeating model or tool calls.diff_run_reports

Why A Runtime Layer Exists​

The protocol models describe what travels through the system: a Task contains messages, parts, state, and artifacts that agents and clients can exchange. An application runtime must additionally decide how that work executes: which run it belongs to, what operation is about to happen, whether that operation is permitted, how approval is obtained, and what progress the user sees.

Without shared runtime primitives, each application tends to invent metadata keys, approval dictionaries, event names, and side-effect checks. Those private conventions work initially, but become difficult to propagate across agents, serialize through transports, test deterministically, or reuse in another interface. Protolink's runtime layer gives those concerns stable contracts while leaving application meaning and presentation outside the framework.

The central lifecycle is:

This lifecycle is not limited to LLM-selected tool calls. The same RunAction and policy contracts can protect deterministic flows and direct application calls.

Runtime Primitives At A Glance​

PrimitiveQuestion it answers
TaskWhat work and results are exchanged between participants?
RunContextWhich run is this, and what constraints travel with it?
CancellationTokenHas live cancellation been requested for active work?
ContextManifestWhat estimated prompt context is about to enter a model?
BudgetPolicy / BudgetEnforcerIs the run still under its configured execution limits?
RunActionWhat concrete operation is about to execute?
ArtifactWhat output or pre-execution preview can be inspected?
PolicyDecisionIs this action allowed, denied, or approval-gated?
ApprovalRequest / ApprovalDecisionWhat must an application approve, and what did it decide?
RunEventWhat is happening now in a stable application-facing format?
EventSinkWhere should normalized runtime events be delivered?

Protolink does not define a universal permission taxonomy, approval screen, or domain-specific action type. Applications choose capability names, build meaningful preview artifacts, and decide whether approval appears in a terminal, desktop UI, web application, editor, or external service. The runtime only guarantees that the decision occurs before execution and that the result is represented consistently.

RunBudget is enforced by the default Agent and LLM inference paths through BudgetEnforcer. The built-in policy allows work under budget, emits warning events near limits, and raises before model or tool execution when a hard limit would be exceeded. The Agent shares one enforcer across the executable parts of a task, including physical provider retries, while nested tasks receive independent scopes. Applications can still provide their own policy when they want compaction, truncation, approval, or domain-specific accounting.

Runtime Context​

RunContext is the typed execution envelope for a task run. It replaces ad hoc metadata keys such as task.metadata["session_id"], trace_id, workspace, or parent_agent with one serializable object stored under task.metadata["run_context"].

Think of the context as information that belongs to the execution but is not the task's business payload. A prompt or record ID belongs in a Message, Part, or action payload; correlation IDs, permissions, cancellation state, and limits belong in RunContext.

from protolink import RunBudget, RunContext, Task

task = Task.create_infer(prompt="Summarize the latest report")

context = RunContext(
    run_id="run_123",
    session_id="session_abc",
    trace_id="trace_abc",
    workspace_uri="file:///workspace",
    agent_chain=["gateway"],
    permissions={"fs.read": {"paths": ["file:///workspace"]}},
    budget=RunBudget(max_steps=8, max_llm_calls=4),
)

context.attach_to_task(task)

The default Agent runtime calls RunContext.ensure_task_context() before normal execution, streaming execution, and outbound agent calls. Existing callers can keep setting task.metadata["session_id"]; Protolink upgrades that legacy metadata into a typed context and mirrors common keys back for compatibility.

Three IDs serve different purposes:

  • run_id identifies one execution attempt and correlates its actions and events.
  • session_id groups related runs, commonly for conversation or application continuity.
  • trace_id correlates observability data and may span several runs or agents.

When work is delegated, RunContext.child() creates a new run identity while preserving the session, trace, workspace, permissions, budget, and application metadata. parent_run_id and agent_chain then describe how execution reached that child.

RunContext API​

dataclassprotolink.RunContext
source
RunContext(
  run_id: str = <generated "run_" ID>,
  session_id: str | None = None,
  trace_id: str | None = None,
  workspace_uri: str | None = None,
  parent_run_id: str | None = None,
  agent_chain: list[str] = [],
  permissions: dict[str, Any] = {},
  budget: RunBudget = RunBudget(),
  canceled: bool = False,
  cancel_reason: str | None = None,
  metadata: dict[str, Any] = {},
  created_at: str = <UTC timestamp>,
)

Mutable, serializable execution metadata for one logical run. Dataclass factory defaults create independent lists, mappings, budgets, IDs, and timestamps for every instance.

Parameters

run_idstrdefault: generated "run_" ID

Stable logical-run identifier. from_task() uses an existing task ID when no typed context or explicit run ID exists.

session_idstr | Nonedefault: None

Conversation or application session shared across related runs.

trace_idstr | Nonedefault: None

Observability correlation ID that may span several runs or agents.

workspace_uristr | Nonedefault: None

Generic execution boundary such as a folder, dataset, browser profile, account, or ticket collection.

parent_run_idstr | Nonedefault: None

Parent logical run for delegated or nested execution.

agent_chainlist[str]default: []

Ordered agents that handled the run.

permissionsdict[str, Any]default: {}

Domain-neutral capability rules or scoped policy metadata. Context rules can narrow, but cannot weaken, the configured runtime policy.

budgetRunBudgetdefault: RunBudget()

Execution limits. The field is always a RunBudget; an unconstrained default has every limit set to None.

canceledbooldefault: False

Serializable cancellation state, separate from the process-local CancellationToken.

cancel_reasonstr | Nonedefault: None

Optional explanation retained with a canceled context.

metadatadict[str, Any]default: {}

Application-owned data that should travel with the run.

created_atstrdefault: current UTC timestamp

ISO timestamp captured at construction.

Serialization and task binding

to_dict()dict[str, Any]

Serializes all fields, including the nested budget.

from_dict(data)RunContext

Accepts a mapping or None, understands legacy workspace, budgets, cancelled, and cancellation_reason spellings, and generates missing identity/time values.

from_task(task, *, default_session_id=None)RunContext

Reads task.metadata["run_context"], merges compatible top-level legacy metadata, and returns a detached mutable context. It does not write back to the task.

ensure_task_context(task, *, default_session_id=None, agent_name=None)RunContext

Normalizes a task context, optionally appends an agent, persists it back to task metadata, and returns it.

attach_to_task(task)None

Mutates task.metadata: stores the complete context under run_context and mirrors populated correlation and cancellation keys at the top level. Existing mirrored keys are not deleted when a field later becomes None.

Copy helpers

with_agent(agent_name)RunContext

Returns a copy with the agent appended unless it is already the final chain entry.

child(*, run_id=None, agent_name=None)RunContext

Returns a new run with parent_run_id=self.run_id, preserving session, trace, workspace, permission, budget, chain, and metadata values.

cancel(reason=None)RunContext

Returns a canceled copy; it does not mutate this context or signal live execution.

copy(**overrides)RunContext

Round-trips through serialization and returns a top-level defensive copy with selected replacements. Nested application values remain ordinary caller-owned objects.

Mutation boundary

RunContext itself is mutable. The with_agent(), child(), cancel(), and copy() helpers return new contexts, while attach_to_task() and ensure_task_context() intentionally mutate task metadata.

RunContext.permissions accepts capability rules using allow, deny, or require_approval. Boolean values are also supported: True allows and False denies. Runtime-owned policy and context rules are combined using the most restrictive result, so task metadata can narrow but cannot weaken the agent's configured policy. RunContext.budget is enforced by the built-in LLM loop for steps, LLM calls, tool calls, runtime seconds, input tokens, and output tokens.

This most-restrictive rule is important at trust boundaries. An incoming task may request fewer privileges for a run, but it cannot grant itself more authority than the receiving agent's policy allows.

RunBudget​

dataclassprotolink.RunBudget
source
RunBudget(
  max_steps: int | None = None,
  max_llm_calls: int | None = None,
  max_tool_calls: int | None = None,
  max_runtime_seconds: float | None = None,
  max_input_tokens: int | None = None,
  max_output_tokens: int | None = None,
  metadata: dict[str, Any] = {},
)

Mutable limit container carried by RunContext. It records policy input; BudgetEnforcer performs the actual checks.

Parameters

max_stepsint | Nonedefault: None

Maximum logical inference/runtime step.

max_llm_callsint | Nonedefault: None

Maximum model calls admitted by one enforcer.

max_tool_callsint | Nonedefault: None

Maximum model-selected tool calls admitted by one enforcer.

max_runtime_secondsfloat | Nonedefault: None

Maximum wall-clock seconds measured from enforcer construction.

max_input_tokensint | Nonedefault: None

Aggregate pre-call input-token limit.

max_output_tokensint | Nonedefault: None

Aggregate output-token limit checked after model usage is known or estimated.

metadatadict[str, Any]default: {}

Application-specific limits or annotations not interpreted by the default policy.

Methods

to_dict()dict[str, Any]

Returns the complete JSON-compatible budget shape.

from_dict(data)RunBudget

Accepts a mapping or None, coerces known numeric fields with int()/float(), and preserves unknown keys inside metadata.

Raises

TypeError | ValueError

Numeric coercion errors from malformed serialized values propagate from from_dict().

Validation

The dataclass does not reject negative or internally inconsistent limits. The default policy compares observed values literally, so applications should construct non-negative budgets.

Context Manifests And Budgets​

Before every LLM call, Protolink prepares a ContextManifest. It is provider-neutral and estimates the context that is about to enter the model: compiled system instructions, runtime affordances such as tools and delegation targets, prior conversation history, and the current user query.

from protolink import ContextManifest, LLMModelProfile, RunBudget, RunContext, create_llm

llm = create_llm("mock")
llm.configure_metrics(LLMModelProfile(context_window=8192))

context = RunContext(
    run_id="run_budgeted",
    budget=RunBudget(max_steps=4, max_llm_calls=2, max_input_tokens=6000),
)

events = []

async def capture(event):
    events.append(event)

await llm.infer(
    query="Summarize this context",
    tools={},
    run_context=context,
    event_callback=capture,
)

manifest = ContextManifest.from_dict(events[1]["manifest"])

ContextItem​

dataclassprotolink.ContextItem
source
ContextItem(
  kind: str,
  name: str,
  tokens: int,
  metadata: dict[str, Any] = {},
)

Immutable token estimate for one logical context section.

Parameters

kindstrrequired

Extensible category such as "system", "tool_prompt", "history", or "user".

namestrrequired

Stable display/test name for the section.

tokensintrequired

Estimated section token count.

metadatadict[str, Any]default: {}

Section-specific details such as message, tool, or delegated-agent counts.

Methods

to_dict()dict[str, Any]

Serializes all fields.

from_dict(data)ContextItem

Supplies fallback names, coerces tokens to an integer when possible, and clamps restored token counts to zero or greater.

ContextManifest​

dataclassprotolink.ContextManifest
source
ContextManifest(
  run_id: str | None = None,
  session_id: str | None = None,
  agent_name: str | None = None,
  provider: str | None = None,
  model: str | None = None,
  system_tokens: int = 0,
  history_tokens: int = 0,
  tool_prompt_tokens: int = 0,
  user_tokens: int = 0,
  context_items: tuple[ContextItem, ...] = (),
  total_estimated_tokens: int = 0,
  context_window: int | None = None,
  estimated: bool = True,
  metadata: dict[str, Any] = {},
  created_at: str = <UTC timestamp>,
)

Immutable provider-neutral preflight summary for one model input.

Correlation and model

run_idstr | Nonedefault: None

Logical run ID copied from RunContext.

session_idstr | Nonedefault: None

Optional session correlation ID.

agent_namestr | Nonedefault: None

Current agent, explicitly supplied or inferred from the final context-chain entry.

providerstr | Nonedefault: None

Provider identifier supplied by the LLM wrapper.

modelstr | Nonedefault: None

Model identifier used by estimation.

Token estimates

system_tokensintdefault: 0

Estimated non-tool system instructions.

history_tokensintdefault: 0

Estimated prior conversation, excluding the newest matching current query.

tool_prompt_tokensintdefault: 0

Estimated tool and delegation declarations included in runtime affordances.

user_tokensintdefault: 0

Estimated current query.

context_itemstuple[ContextItem, ...]default: ()

Per-section records for interfaces and assertions.

total_estimated_tokensintdefault: 0

Additive estimate used for pre-call input-budget checks.

context_windowint | Nonedefault: None

Optional window copied from LLMModelProfile; no overflow decision is made by this dataclass.

estimatedbooldefault: True

Indicates that the counts are estimates rather than provider-reported usage.

Metadata and serialization

metadatadict[str, Any]default: {}

Extensible manifest details.

created_atstrdefault: current UTC timestamp

ISO construction time.

to_dict(*, redaction_policy=None)dict[str, Any]

Serializes the manifest and optionally applies recursive RedactionPolicy masking.

from_dict(data)ContextManifest

Restores items, coerces numeric fields when possible, clamps token counts to non-negative values, and regenerates a missing timestamp.

build_context_manifest​

functionprotolink.build_context_manifest
source
build_context_manifest(
  *,
  history: ConversationHistory,
  query: str,
  run_context: RunContext | None = None,
  agent_name: str | None = None,
  provider: str | None = None,
  model: str | None = None,
  profile: LLMModelProfile | None = None,
  tools: dict[str, Any] | None = None,
  agent_cards: list[Any] | None = None,
) -> ContextManifest

Builds the manifest used immediately before an LLM call without changing conversation history.

Parameters

historyConversationHistoryrequired

Prepared conversation history, including the compiled system prompt.

querystrrequired

Current user query. The newest equal user message is excluded from history and counted here instead.

run_contextRunContext | Nonedefault: None

Supplies run/session IDs and a fallback agent name.

agent_namestr | Nonedefault: None

Explicit current agent, taking precedence over the context chain.

providerstr | Nonedefault: None

Optional provider label.

modelstr | Nonedefault: None

Optional model identifier passed to token estimation.

profileLLMModelProfile | Nonedefault: None

Supplies only context_window to the returned manifest.

toolsdict[str, Any] | Nonedefault: None

Exposed tools summarized by name, description, input schema, and capabilities.

agent_cardslist[Any] | Nonedefault: None

Delegation targets included in runtime-affordance estimation.

Returns

manifestContextManifest

A new immutable estimate with system, tool/delegation, history, user, total, and per-section counts.

Estimation

Tool declarations are estimated as a separate descriptor payload and then subtracted from raw system tokens, clamped at zero. Counts are useful for consistent budgeting but are not provider billing records.

BudgetEnforcer applies RunBudget during task execution and inference:

LimitEnforcement point
max_stepsBefore each explicit tool part and each inference step begins.
max_llm_callsBefore every physical provider attempt, including transient retries.
max_tool_callsBefore an explicit or model-selected tool executes.
max_input_tokensBefore every provider attempt, using the current ContextManifest.
max_output_tokensAfter provider usage or local output estimates are available.
max_runtime_secondsAt every preflight and after provider calls; completed tool/delegation results are recorded before the next preflight.

Warnings are emitted as budget.warning; hard denials are emitted as budget.exceeded and raise BudgetExceededError before the protected operation proceeds. A provider-call runtime or output-token denial can occur after the request completes. Tool and delegation side effects are not treated as reversible: after they return, their result is recorded and the next preflight prevents additional work.

BudgetUsage​

dataclassprotolink.BudgetUsage
source
BudgetUsage(
  steps: int = 0,
  llm_calls: int = 0,
  tool_calls: int = 0,
  input_tokens: int = 0,
  output_tokens: int = 0,
  runtime_seconds: float = 0.0,
  metadata: dict[str, Any] = {},
)

Immutable usage snapshot evaluated against a RunBudget.

Fields

stepsintdefault: 0
Cumulative admitted runtime steps for this enforcer.
llm_callsintdefault: 0
Model calls admitted by this enforcer.
tool_callsintdefault: 0
Tool calls admitted by this enforcer.
input_tokensintdefault: 0
Aggregate pre-call input tokens.
output_tokensintdefault: 0
Aggregate observed output tokens.
runtime_secondsfloatdefault: 0.0
Elapsed wall-clock time.
metadatadict[str, Any]default: {}
Application-owned counters.

Methods

to_dict()dict[str, Any]
Serializes every field.
from_dict(data)BudgetUsage

Accepts a mapping or None; malformed known numeric values become zero rather than raising.

BudgetDecision​

dataclassprotolink.BudgetDecision
source
BudgetDecision(
  effect: BudgetDecisionEffect = "allow",
  limit_name: str | None = None,
  observed: int | float | None = None,
  limit: int | float | None = None,
  message: str | None = None,
  usage: BudgetUsage | None = None,
  metadata: dict[str, Any] = {},
  timestamp: str = <UTC timestamp>,
)

Immutable policy result. BudgetDecisionEffect accepts "allow", "warn", "deny", "compact", "truncate", or "require_approval"; the built-in policy emits only the first three.

Fields

effectBudgetDecisionEffectdefault: "allow"

Requested control outcome.

limit_namestr | Nonedefault: None

RunBudget field responsible for the decision.

observedint | float | Nonedefault: None

Current or projected usage.

limitint | float | Nonedefault: None

Configured hard limit.

messagestr | Nonedefault: None

Human-readable event/error text.

usageBudgetUsage | Nonedefault: None

Full evaluated snapshot.

metadatadict[str, Any]default: {}

Application policy details.

timestampstrdefault: current UTC timestamp

Decision creation time.

Properties and methods

allowedbool

True only for "allow" and "warn"; custom compact/truncate/approval effects require application handling.

to_dict()dict[str, Any]

Serializes the decision and nested usage.

allow(usage)BudgetDecision

Class method returning a standard allow decision.

BudgetPolicy​

classprotolink.BudgetPolicy
source
BudgetPolicy(
  *,
  warning_ratio: float = 0.8,
)

Deterministic comparison policy for configured hard limits.

Parameters

warning_ratiofloatdefault: 0.8

Fraction at or above which a configured positive limit warns. 0 disables warnings.

Methods

evaluate(budget, usage)BudgetDecision

Returns the first hard denial where observed > limit; otherwise returns the first warning where observed >= limit * warning_ratio; otherwise allows. Equality with a hard limit is permitted.

Raises

ValueError

Construction rejects a negative warning_ratio.

BudgetEnforcer​

classprotolink.BudgetEnforcer
source
BudgetEnforcer(
  context_or_budget: RunContext | RunBudget | None = None,
  *,
  policy: BudgetPolicy | None = None,
)

Stateful counter and wall-clock tracker used by task execution and the inference loop. Direct callers normally create one per run; Agent binds one per task ID and restores an outer scope after inline nested work.

Parameters

context_or_budgetRunContext | RunBudget | Nonedefault: None

Supplies limits directly or through a context. None uses an unconstrained budget.

policyBudgetPolicy | Nonedefault: None

Evaluation policy; None creates the default BudgetPolicy.

Attributes

budgetRunBudget
Effective limit object.
policyBudgetPolicy
Effective policy.
usageBudgetUsage
Latest committed allowed/warned usage.
has_output_token_limitbool

Whether a post-call output-token check is needed.

Checks

check_step(step)BudgetDecision

Projects steps=step, measures elapsed runtime, and commits the snapshot only when the decision is allowed.

check_next_step()BudgetDecision

Increments from the currently committed step count. Use this when several infer or tool operations share one enforcer and no caller-local step number is authoritative.

check_llm_call(*, input_tokens=0)BudgetDecision

Projects one additional model call and non-negative input tokens before execution.

check_tool_call()BudgetDecision

Projects one additional tool call before execution.

record_output_tokens(output_tokens)BudgetDecision

Adds non-negative tokens after a model call. None returns an allow decision without invoking the policy.

evaluate()BudgetDecision

Evaluates current counters with refreshed elapsed runtime without committing a new snapshot.

Denials and warnings

These methods return decisions; they do not raise BudgetExceededError themselves. The default inference integration emits events and raises from a deny decision. Denied projections are not committed, and each warning limit is surfaced only once per enforcer.

BudgetExceededError​

exceptionprotolink.BudgetExceededError
source
BudgetExceededError(
  decision: BudgetDecision,
)

Runtime error carrying the denying BudgetDecision on its decision attribute. Its message is decision.message or "Run budget exceeded".

Parameters

decisionBudgetDecisionrequired

Denying decision retained on the exception and used to construct its message.

Canceling Running Tasks​

Protolink distinguishes cancellation state from live cancellation control:

  • Task.cancel() changes the serializable protocol state to canceled.
  • RunContext.cancel() creates a serializable canceled context snapshot.
  • CancellationToken signals process-local code that active execution must stop.
  • The Agent's active-task registry connects a task ID to its token and owning asyncio.Task while that task is running.

This separation keeps Task and RunContext safe to send through transports while allowing the runtime to interrupt an actual coroutine. A Python synchronization object is never placed in task metadata or sent to another agent.

Cancellation Lifecycle​

The task ID is available before submission because Protolink tasks are created by the caller. A CLI or UI can therefore keep the ID associated with a running operation and issue cancellation from another coroutine or control request.

Direct Agent Cancellation​

import asyncio

from protolink import Agent, AgentCard, Task

agent = Agent(AgentCard(name="worker", description="Worker", url="runtime://worker"))
task = Task.create_infer(prompt="Perform long-running work")

running = asyncio.create_task(agent.run_task(task))
# Cancellation targets active execution, so wait until registration completes.
while task.id not in agent.active_task_ids:
    await asyncio.sleep(0)
canceled = await agent.cancel_task(task.id, reason="Stopped by user")
result = await running

assert canceled.state.value == "canceled"
assert result.state.value == "canceled"

The default handle_task() path also registers direct calls through execute_task(). run_task() is the server-facing wrapper and should be used by direct callers that override handle_task() completely, because it guarantees active-task registration around custom logic.

Remote Cancellation​

task = Task.create_infer(prompt="Perform long-running work")
running = asyncio.create_task(client.send_task(agent_url, task))

# In a real application, enable the cancel control after the first status or
# progress event confirms that the remote agent accepted the task.
await task_started.wait()

canceled = await client.cancel_task(
    agent_url,
    task.id,
    reason="Stopped from the application",
)
result = await running

AgentClient.cancel_task() uses ProtoLink's native POST /tasks/cancel operation and returns the updated task. The HTTP adapter exposes the canonical A2A 1.0 CancelTask operation separately. The native client call works over HTTP, SSE JSON-RPC, WebSocket, gRPC, and RuntimeTransport; WebSocket uses a separate control connection so cancellation cannot wait behind the request or stream it needs to stop.

The synchronous client exposes the same operation as client.sync.cancel_task(...). A synchronous call can only cancel work running on another thread, process, or event loop; it cannot interrupt itself while blocked in the same call stack.

Cooperative Checkpoints​

The default runtime checks the token:

  • before each task part;
  • before inference starts and at every inference step;
  • before model-selected tools and delegated-agent calls;
  • before authorization and dispatch; and
  • before every physical provider attempt, including a retry.

If cancellation interrupts an awaited operation, normal coroutine cancellation applies. If the operation wins the race and returns first, ProtoLink records its result rather than discarding evidence of a potentially committed side effect; cancellation is honored before any subsequent work.

The registry also calls asyncio.Task.cancel(), so an async model request, async tool, retry sleep, or delegated call normally stops immediately at its current await point. Custom handlers can retrieve the live token with agent.get_cancellation_token(task.id) and call token.raise_if_cancelled() inside CPU loops or between application-defined stages.

Cancellation of a parent model-driven delegation schedules a best-effort cancellation request for the child task. The child receives its own RunContext with parent_run_id, preserving trace and run relationships.

Final State And Events​

Successful cancellation synchronizes all application-visible surfaces:

  • Task.state becomes canceled and task.metadata["cancel_reason"] is set.
  • RunContext.canceled becomes True and carries the same reason.
  • Streaming finishes with one final task.status / TaskStatusUpdateEvent whose state is canceled.
  • Cancellation is not emitted as task.error and is not converted to failed.
  • The active registry entry is removed in finally, including after errors and cancellation.

A protocol cancellation requested through Agent.cancel_task() is consumed by the execution wrapper and returned as the canceled task or final canceled stream event. If the owning coroutine is canceled externally, ProtoLink still marks and persists the task but re-raises asyncio.CancelledError so normal asyncio cancellation semantics are preserved. Closing a task stream before its terminal event also marks and persists unfinished work as canceled rather than leaving an orphaned working task.

Cancellation does not erase an operation that already returned. Explicit tool results are preserved; late cancellation is described by task.metadata["completed_after_cancellation"], and runtime overruns are appended to task.metadata["completed_action_budget_overruns"]. Successful model-selected tools and delegations receive immediately snapshotted Artifact(kind="action_result") JSON receipts carrying completion status, action_id, source="inference", action kind, and inference step. The internal result is intentionally omitted from this client-visible receipt and remains in private LLM history. Each completed top-level task part is attached and snapshotted immediately, so a later part failure retains earlier progress.

Requests for unknown active IDs raise TaskNotFoundError. This includes a cancellation request that arrives before task registration or after cleanup, so applications should wait for task acceptance or the first streamed status before enabling a cancel control. A task still registered but already terminal raises TaskNotCancelableError. The registry contains active execution only; durable lookup of completed tasks belongs in application storage.

Best-Effort Guarantees​

Cancellation cannot safely promise that every external operation has stopped:

  • Async Python work is interruptible when it reaches an await or explicit token checkpoint.
  • A synchronous function running on the event-loop thread cannot process cancellation until it returns.
  • Moving synchronous work to a thread keeps the event loop responsive, but Python cannot forcibly terminate that thread.
  • A model provider, database, subprocess, or remote API may continue work after the local request is abandoned.
  • Destructive operations should place their commit as late as possible, check cancellation beforehand, or use a subprocess/service that supports its own cancellation or rollback protocol.

For this reason, Protolink follows A2A's best-effort model: it attempts cancellation and reports the resulting task state, while tools and external systems remain responsible for stronger transactional guarantees.

TaskCancellationRequest​

dataclassprotolink.TaskCancellationRequest
source
TaskCancellationRequest(
  id: str,
  reason: str | None = None,
  metadata: dict[str, Any] = {},
)

Immutable A2A-compatible task-ID control payload.

Parameters

idstrrequired

Active task ID. Whitespace-only values are rejected.

reasonstr | Nonedefault: None

Optional human-readable cancellation reason.

metadatadict[str, Any]default: {}

Additional control-plane data. Construction defensively copies the mapping and inserts reason only when that key is absent.

Methods

to_dict()dict[str, Any]

Returns {"id": ..., "metadata": ...}; the reason lives inside metadata to match task-ID parameter wire shapes.

from_dict(data)TaskCancellationRequest

Accepts id or legacy task_id; a top-level reason takes precedence over metadata["reason"].

Raises

ValueError

Raised when the resolved task ID is empty or whitespace-only.

CancellationToken​

classprotolink.CancellationToken
source
CancellationToken()

Thread-safe, process-local cooperative signal backed by threading.Event, allowing cancellation to cross event-loop threads without entering serialized task metadata.

Properties

is_cancelledbool

Whether the first cancellation request has been recorded.

reasonstr | None

First supplied reason, protected by the token lock.

canceled_atstr | None

UTC timestamp of the first request.

Methods

cancel(reason=None)bool

Mutates the token exactly once. Returns True for the first request and False thereafter, preserving the first reason and timestamp.

raise_if_cancelled()None

Returns normally while active; after cancellation raises asyncio.CancelledError with the reason or a default message.

Raises

asyncio.CancelledError

Raised by raise_if_cancelled() after signaling. It is a cancellation control exception, so broad except Exception blocks may not catch it.

Cancellation errors​

exception familyprotolink.TaskCancellationError
source
TaskCancellationError(message: str)

Control-plane failures share a RuntimeError base and contain no additional structured attributes.

Parameters

messagestrrequired

Human-readable failure message passed to RuntimeError.

Subclasses

TaskNotFoundErrorTaskCancellationError

The requested ID is not in the active execution registry, including requests before registration or after cleanup.

TaskNotCancelableErrorTaskCancellationError

A known active record already contains a terminal task.

TaskAlreadyRunningErrorTaskCancellationError

Another coroutine attempts to register the same task ID concurrently; nested registration by the owning coroutine is allowed.

Runtime Actions And Artifacts​

RunAction is the concrete operation that Protolink evaluates immediately before a side effect. It is separate from provider or LLM action formats, so deterministic flows and direct callers use the same contract.

An LLM action is a planning output: it says that a model wants to call a tool or delegate work. A RunAction is the runtime's prepared execution record after the target is known and arguments have been validated. Policy evaluates the latter because it is the closest reliable description of what will actually happen.

from protolink import Artifact, Part, RunAction

preview = Artifact(
    kind="preview",
    name="record update",
    media_type="application/json",
    parts=[Part.json({"record_id": "42", "status": "published"})],
)

action = RunAction(
    kind="resource.update",
    name="publish_record",
    payload={"arguments": {"record_id": "42"}},
    capabilities=frozenset({"records.write"}),
).with_artifacts([preview])

Every action has a stable action_id, an extensible kind, structured payload, required capabilities, and optional preview or result artifacts. Artifact descriptors now include kind, name, uri, media_type, and action_id while retaining their existing parts and metadata fields.

Applications can use preview artifacts for any operation that benefits from inspection before execution: a resource update, outbound message, database mutation, browser action, generated file, or domain-specific command.

Artifacts attached before execution are descriptive; they do not perform the operation. This makes them safe to render in an approval interface. The actual side effect remains inside the tool or application executor and only runs after authorization succeeds.

RunAction​

dataclassprotolink.RunAction
source
RunAction(
  kind: str,
  name: str,
  payload: dict[str, Any] = {},
  capabilities: frozenset[str] = frozenset(),
  artifacts: tuple[Artifact, ...] = (),
  description: str | None = None,
  metadata: dict[str, Any] = {},
  action_id: str = <generated "action_" ID>,
  created_at: str = <UTC timestamp>,
)

Immutable prepared side-effect intent evaluated immediately before execution. The dataclass is frozen, although caller-supplied payload and metadata can themselves contain mutable values.

Parameters

kindstrrequired

Non-empty extensible category such as "tool.call", "agent.call", or an application-defined operation.

namestrrequired

Non-empty operation or target name.

payloaddict[str, Any]default: {}

Validated structured input; tool actions conventionally use an "arguments" member.

capabilitiesfrozenset[str]default: frozenset()

Required application-defined authorities. Construction normalizes any supplied iterable into non-empty strings and a frozenset.

artifactstuple[Artifact, ...]default: ()

Preview or result descriptors normalized to a tuple.

descriptionstr | Nonedefault: None

Optional concise explanation for approval interfaces and logs.

metadatadict[str, Any]default: {}

Extensible runtime/application data.

action_idstrdefault: generated "action_" ID

Stable correlation ID for events and artifacts.

created_atstrdefault: current UTC timestamp

Preparation timestamp.

Copy helpers

with_artifacts(artifacts)RunAction

Returns a new action. Artifacts missing action_id are copied and correlated to this action; pre-existing IDs are preserved.

with_capabilities(capabilities)RunAction

Returns a new action requiring the union of existing and supplied non-empty string capabilities.

with_payload(payload)RunAction

Returns a new action with a top-level defensive copy of the replacement mapping.

Serialization and errors

to_dict()dict[str, Any]

Serializes sorted capabilities and nested artifacts.

from_dict(data)RunAction

Restores nested artifacts and supplies "action"/"unnamed" plus generated identity/time values when omitted.

ValueError

Direct construction rejects a blank kind or name.

Artifact​

dataclassprotolink.Artifact
source
Artifact(
  id: str = <generated artifact ID>,
  parts: list[Part] = [],
  metadata: dict[str, Any] = {},
  timestamp: str = <UTC timestamp>,
  kind: str = "result",
  name: str | None = None,
  uri: str | None = None,
  media_type: str | None = None,
  action_id: str | None = None,
)

Mutable structured output or pre-execution preview.

Parameters

idstrdefault: generated artifact ID
Stable artifact identity.
partslist[Part]default: []
Ordered content parts.
metadatadict[str, Any]default: {}
Application-owned details.
timestampstrdefault: current UTC timestamp
Creation time.
kindstrdefault: "result"
Extensible category such as "result", "preview", or "diagnostic".
namestr | Nonedefault: None
Optional display/resource name.
uristr | Nonedefault: None
Optional represented-resource URI.
media_typestr | Nonedefault: None
Optional artifact-level MIME type.
action_idstr | Nonedefault: None
Related RunAction identity.

Mutation and serialization

add_part(part)Artifact

Appends the exact Part to parts, mutates this artifact, and returns self.

add_text(text)Artifact

Creates and appends Part.text(text), mutates this artifact, and returns self.

for_action(action_id)Artifact

Replaces action_id, mutates this artifact, and returns self.

to_dict()dict[str, Any]

Serializes nested parts and descriptors.

from_dict(data)Artifact

Restores nested parts and remains compatible with older payloads lacking descriptor fields.

Capability Policy​

Tools declare the capabilities they require. CapabilityPolicy supports exact rules and namespace wildcards such as records.*. The strongest result wins across all capabilities required by an action: deny outranks require_approval, which outranks allow.

Capabilities describe authority, not tool implementation. A tool named publish_record might require records.write; another application could use messages.send, browser.navigate, or inventory.adjust. Protolink treats these as opaque names and only applies the configured rules.

from protolink import Agent, ApprovalDecision, CapabilityPolicy

policy = CapabilityPolicy(
    {
        "records.read": "allow",
        "records.write": "require_approval",
        "records.delete": "deny",
    }
)

async def approve(request, context):
    # Render request.action and request.action.artifacts in any UI.
    return ApprovalDecision(
        approved=True,
        request_id=request.request_id,
        decided_by="operator",
    )

agent = Agent(card, policy=policy, approval_handler=approve)

The built-in policy defaults to allow for backward compatibility. A protected capability is enforced when a tool declares it, a policy rule targets it, or RunContext.permissions restricts it. Applications that need resource-level checks can implement the asynchronous Policy.evaluate(action, context) protocol and inspect the complete action payload.

Use the built-in policy when capability names are sufficient. Implement a custom policy when authorization depends on values such as a resource URI, account, time window, tenant, data classification, or the contents of a preview artifact.

PolicyEffect and Policy​

public typesprotolink.PolicyEffect · protolink.Policy
source
class PolicyEffect(str, Enum):
  ALLOW = "allow"
  DENY = "deny"
  REQUIRE_APPROVAL = "require_approval"

class Policy(Protocol):
  async def evaluate(
      self,
      action: RunAction,
      context: RunContext,
  ) -> PolicyDecision: ...

PolicyEffect is the serialized decision vocabulary. Policy is the structural async contract accepted by ActionAuthorizer; custom implementations need not subclass it.

Policy.evaluate parameters

actionRunActionrequired

Fully prepared operation; evaluation must not execute it.

contextRunContextrequired

Cancellation, permission, identity, and application metadata for the run.

Returns

decisionPolicyDecision

Typed allow, deny, or approval requirement.

PolicyDecision​

dataclassprotolink.PolicyDecision
source
PolicyDecision(
  effect: PolicyEffect,
  reason: str,
  policy_name: str,
  matched_capabilities: tuple[str, ...] = (),
  metadata: dict[str, Any] = {},
)

Immutable serializable result from a runtime policy.

Parameters

effectPolicyEffectrequired

Direct construction also accepts supported strings, booleans, and effect mappings through the same coercion used for capability rules.

reasonstrrequired

Concise explanation.

policy_namestrrequired

Stable producer name for traces and interfaces.

matched_capabilitiestuple[str, ...]default: ()

Capabilities responsible for the strongest effect; iterable input is normalized to a tuple.

metadatadict[str, Any]default: {}

Policy-specific details, including per-capability decisions in the built-in policy.

Methods and errors

to_dict()dict[str, Any]

Serializes the enum to its string value and capabilities to a list.

from_dict(data)PolicyDecision

Restores a decision and defaults missing serialized decisions to deny.

ValueError

Raised when an effect cannot be coerced to allow, deny, or require approval.

CapabilityPolicy​

classprotolink.CapabilityPolicy
source
CapabilityPolicy(
  rules: Mapping[str, PolicyEffect | str | bool | Mapping[str, Any]] | None = None,
  *,
  default_effect: PolicyEffect | str = PolicyEffect.ALLOW,
  name: str = "capability_policy",
)

First-party capability matcher. Runtime and context rules are combined per required capability, then the most restrictive combined result wins.

Parameters

rulesMapping[str, PolicyEffect | str | bool | Mapping[str, Any]] | Nonedefault: None

Exact rules, "*" fallback, and namespace wildcards such as "records.*". The longest matching wildcard wins after exact lookup.

default_effectPolicyEffect | strdefault: PolicyEffect.ALLOW

Runtime result for a capability unmatched by rules.

namestrdefault: "capability_policy"

Name embedded in decisions and serialized configuration.

Rule values

effect string or PolicyEffectrule

Accepts allow/allowed, deny/denied, and approval aliases including approval, approve, ask, and require_approval.

boolrule

True means allow and False means deny.

mappingrule

Reads effect, decision, or mode; a mapping with none of those keys is treated as an allowed scoped grant.

Methods

evaluate(action, context)Awaitable[PolicyDecision]

Denies a canceled context first. An action with no capabilities is allowed. Otherwise combines policy and context values using deny > approval > allow and reports the capabilities producing the strongest result.

to_dict()dict[str, Any]

Serializes first-party declarative configuration only, validating nested values and finite numbers.

from_dict(data)CapabilityPolicy

Restores only "type": "capability" data and rejects executable or malformed values.

Raises

ValueError

Unsupported effects, serialized policy types, rule shapes, or invalid names.

TypeError

Non-string capability keys or non-declarative nested configuration.

Mutation

Construction copies the top-level rules mapping, but rules, default_effect, and name remain public mutable attributes. Treat a configured policy as stable while actions are executing.

Approval Checkpoints​

When policy returns require_approval, ActionAuthorizer creates an ApprovalRequest and calls the application-owned approval handler. Protolink controls whether execution may continue; the application controls terminal, desktop, web, service, or editor presentation.

The handler returns an ApprovalDecision correlated by request_id. A denied decision raises ActionDeniedError. If no handler is configured, Protolink fails closed with ApprovalRequiredError, which carries the serializable request.

The handler receives the complete RunAction, including validated arguments, required capabilities, description, metadata, and preview artifacts. It can therefore present useful context without rediscovering the intended operation from raw model output or tool arguments. Returning a decision is the only way an approval-gated action proceeds.

Native tools can attach action previews through action_builder:

from protolink import Artifact, Part, RunAction

def build_preview(arguments, context):
    return RunAction(
        kind="tool.call",
        name="publish_record",
        payload={"arguments": arguments},
        artifacts=(
            Artifact(
                kind="preview",
                name="publication preview",
                parts=[Part.json(arguments)],
            ),
        ),
    )

@agent.tool(
    name="publish_record",
    description="Publish a record",
    capabilities=["records.write"],
    action_builder=build_preview,
)
async def publish_record(record_id: str) -> dict:
    return {"record_id": record_id, "status": "published"}

Tool arguments are validated before the action is prepared and again before execution. Tool-declared capabilities are always merged into a custom action, so an action_builder cannot accidentally omit a required policy check.

For deterministic code that invokes a tool without a Task, use agent.call_tool_in_context(tool_name, context, **arguments). It applies the same argument preparation, capability policy, and approval handler as model-driven execution.

ApprovalRequest​

dataclassprotolink.ApprovalRequest
source
ApprovalRequest(
  action: RunAction,
  policy_decision: PolicyDecision,
  run_id: str,
  request_id: str = <generated "approval_" ID>,
  created_at: str = <UTC timestamp>,
  metadata: dict[str, Any] = {},
)

Immutable checkpoint passed to an application approval handler.

Parameters

actionRunActionrequired

Fully prepared operation, including validated arguments, capabilities, and preview artifacts.

policy_decisionPolicyDecisionrequired

Approval-requiring policy result.

run_idstrrequired

Logical run correlated with the checkpoint.

request_idstrdefault: generated "approval_" ID

Correlation key that the returned decision must reproduce.

created_atstrdefault: current UTC timestamp

Checkpoint creation time.

metadatadict[str, Any]default: {}

Application-owned presentation/service data.

Methods

to_dict(*, redaction_policy=None)dict[str, Any]

Serializes the nested action and decision, optionally masking secret-bearing keys recursively.

from_dict(data)ApprovalRequest

Restores nested typed records and generates missing request/time values.

ApprovalDecision​

dataclassprotolink.ApprovalDecision
source
ApprovalDecision(
  approved: bool,
  request_id: str,
  reason: str | None = None,
  decided_by: str | None = None,
  metadata: dict[str, Any] = {},
  decided_at: str = <UTC timestamp>,
)

Immutable application response to exactly one checkpoint.

Parameters

approvedboolrequired

Whether execution may continue.

request_idstrrequired

Must equal the corresponding request ID when returned to ActionAuthorizer.

reasonstr | Nonedefault: None

Optional explanation; a denied reason becomes part of ActionDeniedError.

decided_bystr | Nonedefault: None

Optional user, service, or policy actor.

metadatadict[str, Any]default: {}

Additional decision data.

decided_atstrdefault: current UTC timestamp

Decision time.

Methods

to_dict()dict[str, Any]
Serializes every field.
from_dict(data)ApprovalDecision

Restores the decision; missing approved fails closed to False, and a missing request ID becomes an empty string.

ApprovalHandler​

protocolprotolink.ApprovalHandler
source
await handler(
  request: ApprovalRequest,
  context: RunContext,
) -> ApprovalDecision | bool

Structural protocol for application-owned approval interfaces. ActionAuthorizer also accepts ordinary synchronous callables with the same parameters.

Parameters

requestApprovalRequestrequired

Complete serializable checkpoint.

contextRunContextrequired

Active run metadata.

Returns

decisionApprovalDecision | bool

A boolean is converted into a correlated ApprovalDecision; an explicit decision must already carry the matching request ID.

ActionAuthorization​

dataclassprotolink.ActionAuthorization
source
ActionAuthorization(
  action: RunAction,
  policy_decision: PolicyDecision,
  approval_request: ApprovalRequest | None = None,
  approval_decision: ApprovalDecision | None = None,
  authorized_at: str = <UTC timestamp>,
)

Immutable proof returned only after policy allows an action or an approver grants its checkpoint.

Fields and methods

actionRunActionrequired
Authorized operation.
policy_decisionPolicyDecisionrequired
Original policy result.
approval_requestApprovalRequest | Nonedefault: None
Checkpoint when approval was required.
approval_decisionApprovalDecision | Nonedefault: None
Granted application response.
authorized_atstrdefault: current UTC timestamp
Authorization completion time.
to_dict()dict[str, Any]
Serializes nested records.
from_dict(data)ActionAuthorization
Restores nested records and a missing timestamp.

ActionAuthorizer​

classprotolink.ActionAuthorizer
source
ActionAuthorizer(
  policy: Policy | None = None,
  approval_handler: ApprovalHandlerLike | None = None,
)

Coordinates the final policy/approval gate; it never executes the action itself.

Parameters

policyPolicy | Nonedefault: None

Async policy; None creates an allow-by-default CapabilityPolicy.

approval_handlerApprovalHandlerLike | Nonedefault: None

Synchronous or asynchronous callable returning ApprovalDecision or bool.

authorize

actionRunActionrequired

Prepared operation passed to policy and, if needed, approval.

contextRunContextrequired

Run metadata passed unchanged to both boundaries.

returnActionAuthorization

Successful typed authorization. Approval records are absent for direct allows and present for approved checkpoints.

Raises

ActionDeniedError

Policy denies or the approval result is false.

ApprovalRequiredError

Policy requires approval but no handler is configured.

TypeError

The handler returns neither a boolean nor an ApprovalDecision.

ValueError

An explicit approval decision carries a different request ID.

policy or handler error

Other exceptions from application policy/approval code propagate unchanged.

Policy errors​

exception familyprotolink.ActionPolicyError
source
ActionPolicyError(
  message: str,
  *,
  action: RunAction,
  decision: PolicyDecision,
)

Structured runtime-policy failures.

Parameters

messagestrrequired

Human-readable exception message passed to RuntimeError.

actionRunActionrequired

Prepared operation that failed authorization.

decisionPolicyDecisionrequired

Policy result responsible for the failure.

Attributes and subclasses

actionRunAction
Prepared operation that did not receive authorization.
decisionPolicyDecision
Policy result responsible for the failure.
ApprovalRequiredError(request)ActionPolicyError

Adds request and is raised when a checkpoint has no configured handler.

ActionDeniedError(*, action, decision, approval_request=None, approval_decision=None)ActionPolicyError

Adds optional checkpoint/decision records. Its message uses the approver's reason when supplied, otherwise the policy reason.

Run Events​

Existing stream events such as TaskStatusUpdateEvent, TaskArtifactUpdateEvent, and TaskLLMStreamEvent remain the transport-compatible event objects. RunEvent is the normalized application-facing envelope for those events.

The distinction lets transports retain backward-compatible event objects while applications consume one versioned shape. A terminal renderer, web client, test recorder, and logging adapter can all switch on the same RunEvent.type values instead of interpreting provider-specific chunks or nested dictionaries.

from protolink import InMemoryEventSink, RunContext

sink = InMemoryEventSink()

async for task_event in agent.handle_task_streaming(task):
    await sink.emit_task_event(task_event, context=RunContext.from_task(task))

events = sink.to_list()

RunEvent​

dataclassprotolink.RunEvent
source
RunEvent(
  type: str,
  run_id: str | None = None,
  task_id: str | None = None,
  agent_name: str | None = None,
  sequence: int | None = None,
  step: int | None = None,
  span_id: str | None = None,
  parent_span_id: str | None = None,
  action_id: str | None = None,
  parent_action_id: str | None = None,
  delegation_id: str | None = None,
  severity: str = "info",
  summary: str | None = None,
  payload: dict[str, Any] = {},
  final: bool = False,
  metadata: dict[str, Any] = {},
  event_id: str = <UUID>,
  version: str = "1.0",
  timestamp: str = <UTC timestamp>,
)

Mutable versioned application-facing envelope for task and inference runtime activity.

Identity and ordering

typestrrequired

Stable type such as task.status, context.prepared, action.requested, or llm.stream. Direct construction does not restrict custom types.

run_idstr | Nonedefault: None
Logical run correlation.
task_idstr | Nonedefault: None
Protocol task correlation.
agent_namestr | Nonedefault: None
Emitter/handler agent.
sequenceint | Nonedefault: None

Monotonic sink order. In-memory sinks assign it only when absent and otherwise preserve caller values.

stepint | Nonedefault: None
Optional runtime or inference step.
event_idstrdefault: generated UUID
Unique envelope identity.
versionstrdefault: "1.0"
Stable envelope version.
timestampstrdefault: current UTC timestamp
Event creation time.

Relationships

span_idstr | Nonedefault: None
Optional causal span.
parent_span_idstr | Nonedefault: None
Optional parent span.
action_idstr | Nonedefault: None
Related runtime action.
parent_action_idstr | Nonedefault: None
Parent action for nested work.
delegation_idstr | Nonedefault: None
Delegated-agent operation.

Presentation and payload

severitystrdefault: "info"

Renderer/log hint. Normalization chooses info, warning, or error; direct callers may use another string.

summarystr | Nonedefault: None
Short progress text.
payloaddict[str, Any]default: {}

Full normalized source payload. Stable runtime metadata is promoted into additional top-level payload keys without removing its original nested representation.

finalbooldefault: False
Whether the source marks a final boundary.
metadatadict[str, Any]default: {}
Envelope-only metadata, including original source type.

Methods

to_dict(*, redaction_policy=None)dict[str, Any]

Serializes the envelope and optionally masks secrets recursively.

from_dict(data)RunEvent

Restores optional numbers with int(), mappings with top-level copies, and generated identity/time/version defaults.

from_task_event(event, *, context=None, sequence=None)RunEvent

Normalizes an event object/dictionary, maps known task and LLM event types, derives severity/summary/relationships, and optionally recovers context from an embedded serialized task payload.

Mutation

RunEvent is mutable so sinks can assign sequence. Serializing or normalizing does not deep-freeze payload and metadata values.

RunEvent.from_task_event(event) can also recover context from an embedded task payload when the event includes a serialized task.

LLM context, budget, and call lifecycle activity is promoted out of raw LLM metadata into stable event types:

Event typeMeaning
context.preparedA ContextManifest was prepared before an LLM call.
llm.call.startedA model call is about to start.
llm.call.completedA model call returned and usage/latency metadata is available.
budget.warningUsage is near a configured RunBudget limit.
budget.exceededA configured budget limit denied further execution.

Action lifecycle activity is also promoted into stable event types:

Event typeMeaning
action.requestedA concrete RunAction is ready for policy evaluation.
action.policyPolicy returned allow, deny, or require approval.
approval.requiredAn ApprovalRequest checkpoint was created.
approval.decidedThe application returned an ApprovalDecision.
action.startedAn authorized tool or agent operation started.
action.completedThe operation completed successfully.
action.denied / action.failedPolicy denied the operation or execution failed.

The promoted manifest, action, request, decision, action_id, parent_action_id, span_id, parent_span_id, and delegation_id values are available directly in RunEvent.payload; the original task stream payload remains intact for compatibility.

Event Sinks​

EventSink is the protocol for consumers of normalized RunEvent objects. InMemoryEventSink is the built-in implementation for tests, local apps, and replay tooling. Use RunRecorder when you also want a durable RunReport after the stream completes.

from protolink import InMemoryEventSink, RunEvent

sink = InMemoryEventSink()
await sink.emit(RunEvent(type="task.progress", summary="Halfway done"))

assert sink.to_list()[0]["sequence"] == 1

Applications can implement their own sinks for terminal rendering, WebSocket fanout, database persistence, or custom observability systems without changing agent execution code.

An event sink observes execution; it does not authorize it. Approval decisions still flow through the configured approval handler, while sinks distribute the resulting lifecycle to interested consumers.

EventSink​

protocolprotolink.EventSink
source
await sink.emit(
  event: RunEvent,
) -> None

Structural async consumer contract. Implementations decide storage, fanout, rendering, or observability behavior; emitting has no authorization meaning.

Parameters

eventRunEventrequired

One already-normalized runtime event.

InMemoryEventSink​

classprotolink.InMemoryEventSink
source
InMemoryEventSink()

Dependency-free process-local event buffer for tests, local interfaces, and recorder tooling.

Attributes

eventstuple[RunEvent, ...]

New immutable tuple view of recorded object references, in insertion order.

Methods

emit(event)Awaitable[None]

Appends the event. If sequence is None, mutates it to the next sequence; explicit sequence values are retained and advance the next counter when needed.

emit_task_event(event, *, context=None)Awaitable[RunEvent]

Calls RunEvent.from_task_event(), records the result, and returns the appended normalized event.

to_list()list[dict[str, Any]]

Serializes all events without automatic redaction.

clear()None

Mutates the sink by removing all events and resetting sequence numbering to one.

Concurrency

The built-in buffer has no lock and is intended for one event-loop/application coordination domain. Use a synchronized sink for cross-thread emitters.

Run Reports, Replay, And Regression Diffing​

RunReport is the durable app-facing summary built from normalized events. It collects context manifests, action records, approval checkpoints, artifacts, LLM metrics, and the final serialized task when the final stream event includes it.

from protolink import (
    RedactionPolicy,
    RunContext,
    RunRecorder,
    RunReplay,
    assert_budget_under,
    assert_no_denied_actions,
    assert_run_events,
)

context = RunContext.from_task(task)
recorder = RunRecorder(context=context)

async for task_event in agent.handle_task_streaming(task):
    await recorder.record_task_event(task_event)

report = recorder.to_report(metadata={"source": "integration-test"})
safe_json = report.to_dict(redaction_policy=RedactionPolicy())

replay = RunReplay(safe_json)
assert_run_events(replay, ["context.prepared", "llm.call.started", "llm.call.completed"])
assert_no_denied_actions(replay)
assert_budget_under(replay, max_total_tokens=8_000)

RunReplay never re-executes tools or model calls. It is a read-only view over report events with helpers such as event_types, iter_events(), and find_events("context.prepared").

RunReport​

dataclassprotolink.RunReport
source
RunReport(
  context: RunContext | None = None,
  context_manifests: tuple[dict[str, Any], ...] = (),
  events: tuple[RunEvent, ...] = (),
  actions: tuple[dict[str, Any], ...] = (),
  approvals: tuple[dict[str, Any], ...] = (),
  artifacts: tuple[dict[str, Any], ...] = (),
  metrics: tuple[dict[str, Any], ...] = (),
  final_task: dict[str, Any] | None = None,
  metadata: dict[str, Any] = {},
  created_at: str = <UTC timestamp>,
)

Immutable report envelope extracted from normalized events. Tuple membership cannot be reassigned, but nested context, events, and dictionaries remain ordinary objects.

Sections

contextRunContext | Nonedefault: None
Optional run metadata.
context_manifeststuple[dict[str, Any], ...]default: ()
Pre-call context snapshots.
eventstuple[RunEvent, ...]default: ()
Chronological normalized events.
actionstuple[dict[str, Any], ...]default: ()
Prepared actions extracted from action/request payloads.
approvalstuple[dict[str, Any], ...]default: ()
Approval required/decided event projections.
artifactstuple[dict[str, Any], ...]default: ()
Artifacts from task.artifact events.
metricstuple[dict[str, Any], ...]default: ()
LLM call metric payloads.
final_taskdict[str, Any] | Nonedefault: None
Serialized task recovered from a final event or supplied explicitly.
metadatadict[str, Any]default: {}
Application-owned report metadata.
created_atstrdefault: current UTC timestamp
Report creation time.

Construction

from_events(events, *, context=None, final_task=None, metadata=None)RunReport

Coerces event mappings, extracts stable sections, deduplicates actions with non-empty action IDs, and uses the newest final event carrying metadata.task when no truthy explicit final task is supplied.

from_dict(data)RunReport

Restores typed context/events and keeps only mapping entries in tuple sections.

Serialization

to_dict(*, redaction_policy=None)dict[str, Any]

Serializes all sections and optionally applies recursive masking.

redacted(policy=None)RunReport

Returns a reconstructed report with the supplied or default redaction policy applied. It does not mutate the source report.

RunRecorder​

classprotolink.RunRecorder
source
RunRecorder(
  *,
  context: RunContext | dict[str, Any] | None = None,
)

In-memory normalized-event recorder that adds report construction to the sink contract.

Parameters and attributes

contextRunContext | dict[str, Any] | Nonedefault: None

Default report/normalization context; serialized mappings are converted at construction.

eventstuple[RunEvent, ...]

Current immutable tuple view from the internal sink.

Recording

emit(event)Awaitable[None]
Records one normalized event.
emit_task_event(event, *, context=None)Awaitable[RunEvent]

Normalizes and records a task event, using the call context before the recorder default.

record_event(event)Awaitable[RunEvent]
Records and returns the same normalized event.
record_task_event(event, *, context=None)Awaitable[RunEvent]
Alias-style normalize/record helper.

Report and mutation

to_report(*, context=None, final_task=None, metadata=None, redaction_policy=None)RunReport

Extracts a new report. An explicit context overrides the recorder default; a redaction policy returns a redacted reconstructed report.

clear()None

Removes events and resets sequence numbering while retaining the recorder's default context.

RunReplay​

classprotolink.RunReplay
source
RunReplay(
  report: RunReport | dict[str, Any] | Iterable[RunEvent | dict[str, Any]],
)

Read-only view that never calls an agent, tool, model, transport, or external service.

Parameters

reportRunReport | dict[str, Any] | Iterable[RunEvent | dict[str, Any]]required

Existing report, serialized full-report mapping, or event iterable. A dictionary is always interpreted as a report mapping rather than a single event.

Properties and methods

reportRunReport
Coerced durable report.
eventstuple[RunEvent, ...]
Recorded-order events.
event_typestuple[str, ...]
Recorded-order types.
iter_events(event_type=None)Iterable[RunEvent]
Lazy iteration over all or matching events.
find_events(event_type)tuple[RunEvent, ...]
Materialized matching events.
assert_events(expected_types, *, ordered=True, allow_extra=True)None

Delegates to assert_run_events().

RedactionPolicy​

dataclassprotolink.RedactionPolicy
source
RedactionPolicy(
  sensitive_keys: frozenset[str] = DEFAULT_SENSITIVE_KEYS,
  replacement: str = "[REDACTED]",
  max_string_length: int | None = None,
)

Immutable recursive masking policy shared by runtime observability objects.

Parameters

sensitive_keysfrozenset[str]default: DEFAULT_SENSITIVE_KEYS

Case-insensitive names normalized by lowercasing and replacing hyphens with underscores. Defaults include API keys, authorization, credentials, passwords, secrets, and tokens.

replacementstrdefault: "[REDACTED]"

Value substituted for a sensitive field's complete value.

max_string_lengthint | Nonedefault: None

Optional maximum non-secret string prefix; truncated strings receive "...".

Methods

is_sensitive_key(key)bool

Matches configured names plus _api_key, _secret, _token, _password, and _credentials suffixes.

redact(value)Any

Converts supported dataclasses/to_dict() objects to JSON-like values and recursively returns masked mappings and containers without mutating the input.

Raises

ValueError

Raised when max_string_length is negative.

Defaults

DEFAULT_REDACTION_POLICY is the shared default instance used by diff formatting and assertion failures. Raw to_dict() methods generally redact only when a policy is explicitly supplied.

Comparing Run Reports​

Execute a baseline and candidate separately, then compare their recorded reports. Final reports are the usual regression input, but the comparison helpers do not inspect or enforce a task lifecycle state:

from protolink import (
    RunReportDiffConfig,
    RunReportTolerance,
    assert_run_matches,
    diff_run_reports,
    normalize_run_report,
)

config = RunReportDiffConfig(
    ignore_paths=("/metadata/build_host",),
    tolerances=(
        # These application-owned scores are 0.910 and 0.915 in the reports.
        RunReportTolerance(
            "/metadata/evaluation_score",
            absolute_tolerance=0.01,
        ),
    ),
)
normalized_baseline = normalize_run_report(baseline_report, config=config)
comparison = diff_run_reports(baseline_report, candidate_report, config=config)

if not comparison.matches:
    print(comparison.format())

# Convenient in a regression test: raises with a formatted, redacted summary.
assert_run_matches(baseline_report, candidate_report, config=config)

The comparison canonicalizes known identifiers, timestamps, and sequence counters in ProtoLink-owned report envelopes. Recognized task-stream events also normalize runtime-derived summaries and timing values. Application-owned values inside tool payloads and report metadata remain exact unless they match an explicit ignore or tolerance rule. The result contains matches, changed_sections, and path-level differences.

RunReportDiffConfig(sections=..., normalize_volatile=True, ignore_paths=(), tolerances=()) controls the comparison. Each RunReportTolerance(path, absolute_tolerance=0.0, relative_tolerance=0.0) allows bounded numeric variation at one selected path; rules are checked in declaration order and the first match wins. An explicit tolerance takes precedence over built-in volatile normalization for the selected numeric value, so a test can opt a timing field back into bounded comparison. Defaults remain strict for fields that are not part of the built-in volatile-field normalization.

Ignore and tolerance paths use RFC 6901 JSON Pointer syntax. * is ProtoLink's extension and matches exactly one path segment, so /metrics/*/usage/total_tokens covers every metric item's token count. ** has no recursive meaning; it is a literal segment. Bracket notation is not interpreted either: /events[0]/type addresses a literal top-level key named events[0], not item zero of events. Use /events/0/type for that list item.

This is comparison, not execution. Neither diff_run_reports() nor assert_run_matches() invokes an agent, model, tool, transport, or external service. For a reproducible regression test, run the candidate against the same input with mock, captured, or otherwise controlled dependencies; with live dependencies, the diff is still useful evidence of what changed but does not make the run deterministic.

Report comparison types​

constants and type aliasesprotolink.ALL_RUN_REPORT_SECTIONS
source
RunReportSection = Literal[
  "context", "context_manifests", "events", "actions", "approvals",
  "artifacts", "metrics", "final_task", "metadata",
]
RunReportDifferenceKind = Literal["added", "removed", "changed"]
RunReportSource = (
  RunReport
  | RunReplay
  | Mapping[str, Any]
  | Iterable[RunEvent | dict[str, Any]]
)
ALL_RUN_REPORT_SECTIONS: tuple[RunReportSection, ...]

Public typing vocabulary and the ordered default projection.

Definitions

RunReportSectionLiteral

Names the nine selectable report sections. Root RunReport.created_at is intentionally outside this projection.

RunReportDifferenceKindLiteral

Structural change category: added, removed, or changed.

RunReportSourceTypeAlias

Accepted report/replay/mapping/event-iterable input. Strings and bytes are explicitly rejected rather than treated as event iterables.

ALL_RUN_REPORT_SECTIONStuple[RunReportSection, ...]

Ordered default: context, context manifests, events, actions, approvals, artifacts, metrics, final task, and metadata.

RunReportTolerance​

dataclassprotolink.RunReportTolerance
source
RunReportTolerance(
  path: str,
  absolute_tolerance: float = 0.0,
  relative_tolerance: float = 0.0,
)

Immutable numeric tolerance for one exact JSON Pointer pattern.

Parameters

pathstrrequired

Non-root RFC 6901 pointer. ProtoLink's * segment matches exactly one key/index; rules are tried in declaration order.

absolute_tolerancefloatdefault: 0.0

Maximum absolute difference, coerced to float.

relative_tolerancefloatdefault: 0.0

Maximum scale-relative difference, coerced to float.

Raises

TypeError | ValueError

The pointer is root/malformed, an RFC 6901 escape is invalid, or either tolerance is negative/non-finite/not float-coercible.

Numeric semantics

Booleans are never treated as numbers. The comparator uses decimal string conversion so arbitrarily large integers do not lose precision.

RunReportDiffConfig​

dataclassprotolink.RunReportDiffConfig
source
RunReportDiffConfig(
  sections: tuple[RunReportSection, ...] = ALL_RUN_REPORT_SECTIONS,
  normalize_volatile: bool = True,
  ignore_paths: tuple[str, ...] = (),
  tolerances: tuple[RunReportTolerance, ...] = (),
)

Immutable normalization/comparison configuration; iterable constructor inputs are normalized to tuples.

Parameters

sectionstuple[RunReportSection, ...]default: ALL_RUN_REPORT_SECTIONS

Ordered unique projection. Unknown and duplicate names are rejected.

normalize_volatilebooldefault: True

Canonicalizes known runtime IDs, timestamps, timing fields, and derived summaries while preserving application-owned values.

ignore_pathstuple[str, ...]default: ()

RFC 6901 pointer patterns whose exact node and complete subtree are omitted.

tolerancestuple[RunReportTolerance, ...]default: ()

Ordered exact-path numeric rules; the first match wins.

Raises

ValueError

Unknown/duplicate sections or malformed ignore paths.

TypeError

A tolerance entry is not a RunReportTolerance.

RunReportDifference​

dataclassprotolink.RunReportDifference
source
RunReportDifference(
  section: RunReportSection,
  path: str,
  kind: RunReportDifferenceKind,
  baseline: Any = <missing>,
  candidate: Any = <missing>,
)

Immutable path-level structural difference. Internal missing sentinels keep an absent value distinct from an explicit None.

Fields and methods

sectionRunReportSectionrequired
Owning top-level section.
pathstrrequired
Escaped RFC 6901 location in the projected report.
kindRunReportDifferenceKindrequired
Added, removed, or changed.
baselineAnydefault: missing
Original value when available.
candidateAnydefault: missing
Candidate value when available.
to_dict()dict[str, Any]

Returns raw values and omits missing sides. It performs no redaction.

RunReportDiff​

dataclassprotolink.RunReportDiff
source
RunReportDiff(
  differences: tuple[RunReportDifference, ...] = (),
  compared_sections: tuple[RunReportSection, ...] = ALL_RUN_REPORT_SECTIONS,
  ignored_paths: tuple[str, ...] = (),
)

Immutable complete comparison result.

Fields and properties

differencestuple[RunReportDifference, ...]default: ()
Path-level changes.
compared_sectionstuple[RunReportSection, ...]default: ALL_RUN_REPORT_SECTIONS
Projection order.
ignored_pathstuple[str, ...]default: ()
Applied ignore patterns.
matchesbool
Whether differences is empty.
changed_sectionstuple[RunReportSection, ...]

Changed section names in compared_sections order.

Methods

to_dict(*, redaction_policy=None)dict[str, Any]

Returns summary fields and differences. Compared values stay raw unless a policy is explicitly supplied.

format(*, max_differences=20, redaction_policy=DEFAULT_REDACTION_POLICY)str

Returns a deterministic terminal/assertion summary, redacted by default. Pass None deliberately to include raw values.

Raises

ValueError

format() rejects a negative max_differences.

normalize_run_report​

functionprotolink.normalize_run_report
source
normalize_run_report(
  source: RunReportSource,
  *,
  config: RunReportDiffConfig | None = None,
) -> dict[str, Any]

Creates a deterministic selected-section projection without mutating the source.

Parameters

sourceRunReportSourcerequired

Report, replay, serialized report mapping, or event iterable.

configRunReportDiffConfig | Nonedefault: None

Projection and normalization rules; None constructs the defaults.

Returns

projectiondict[str, Any]

New JSON-compatible selected-section mapping with ignored nodes removed and recognized volatile values canonicalized.

Raises

TypeError

A source is a string/bytes or cannot be interpreted as one supported shape.

diff_run_reports​

functionprotolink.diff_run_reports
source
diff_run_reports(
  baseline: RunReportSource,
  candidate: RunReportSource,
  *,
  config: RunReportDiffConfig | None = None,
) -> RunReportDiff

Normalizes baseline and candidate independently, then computes the complete structured comparison.

Parameters

baselineRunReportSourcerequired
Expected report or events.
candidateRunReportSourcerequired
Observed report or events.
configRunReportDiffConfig | Nonedefault: None
Shared projection/comparison rules.

Returns

differenceRunReportDiff

All added, removed, and changed paths. Known sequence-like report sections use semantic alignment; ordinary lists compare positionally.

No execution

Comparison performs no agent, model, tool, transport, or external calls.

assert_run_matches​

assertion functionprotolink.assert_run_matches
source
assert_run_matches(
  baseline: RunReportSource,
  candidate: RunReportSource,
  *,
  config: RunReportDiffConfig | None = None,
) -> RunReportDiff

Parameters

baselineRunReportSourcerequired
Expected report or events.
candidateRunReportSourcerequired
Observed report or events.
configRunReportDiffConfig | Nonedefault: None
Normalization/comparison rules.

Returns

differenceRunReportDiff
Successful matching structured result.

Raises

AssertionError

Reports differ. The message comes from default-redacted RunReportDiff.format().

Core serialization is intentionally explicit about secrets. RunReportDifference.to_dict() always returns its raw fields, and RunReportDiff.to_dict() returns raw compared values unless a policy is supplied. Pass redaction_policy=RedactionPolicy() to RunReportDiff.to_dict() before exporting it. RunReportDiff.format() and the assert_run_matches() failure message apply the default redaction policy unless explicitly disabled. The protolink run diff text and JSON views also redact difference values by default.

The assertion helpers are intentionally small:

assert_run_events​

assertion functionprotolink.assert_run_events
source
assert_run_events(
  source: RunReport | RunReplay | Iterable[RunEvent | dict[str, Any]],
  expected_types: Sequence[str],
  *,
  ordered: bool = True,
  allow_extra: bool = True,
) -> None

Parameters

sourceRunReport | RunReplay | Iterable[RunEvent | dict[str, Any]]required

Recorded report/replay or event iterable.

expected_typesSequence[str]required

Event types to require.

orderedbooldefault: True

Requires declaration order when true.

allow_extrabooldefault: True

Allows unlisted observed events.

Matching modes

ordered=True, allow_extra=Trueordered subsequence
Default additive-event-friendly mode.
ordered=True, allow_extra=Falseexact tuple
Requires exact order and count.
ordered=False, allow_extra=Truemembership
Requires every expected type to appear at least once; duplicate expectations do not require duplicate observations.
ordered=False, allow_extra=Falseexact multiset
Requires equal per-type counts regardless of order.

Raises

AssertionError
The selected matching rule fails.

assert_no_denied_actions​

assertion functionprotolink.assert_no_denied_actions
source
assert_no_denied_actions(
  source: RunReport | RunReplay | Iterable[RunEvent | dict[str, Any]],
) -> None

Fails when it observes an action.denied event, an action.policy decision with effect == "deny", or an approval.decided event whose decision has approved is False.

Parameters

sourceRunReport | RunReplay | Iterable[RunEvent | dict[str, Any]]required

Recorded events to inspect.

Raises

AssertionError

Includes labels built from the denied event type and action ID or event ID.

assert_budget_under​

assertion functionprotolink.assert_budget_under
source
assert_budget_under(
  source: RunReport | RunReplay | Iterable[RunEvent | dict[str, Any]],
  *,
  max_input_tokens: int | None = None,
  max_output_tokens: int | None = None,
  max_total_tokens: int | None = None,
  max_runtime_seconds: float | None = None,
) -> dict[str, int | float]

Aggregates recorded usage and checks caller-supplied regression limits.

Parameters

sourceRunReport | RunReplay | Iterable[RunEvent | dict[str, Any]]required

Report or events from which a temporary report can be built.

max_input_tokensint | Nonedefault: None
Optional aggregate input ceiling.
max_output_tokensint | Nonedefault: None
Optional aggregate output ceiling.
max_total_tokensint | Nonedefault: None
Optional aggregate total ceiling.
max_runtime_secondsfloat | Nonedefault: None
Optional summed LLM-latency ceiling.

Returns

usagedict[str, int | float]

input_tokens, output_tokens, total_tokens, and rounded runtime_seconds.

Raises

AssertionError

One or more observed values are strictly greater than their supplied limit; equality passes.

Aggregation

Provider metric usage is summed first. If aggregate input is zero, manifest total_estimated_tokens values are used; if aggregate total is zero, input plus output is used. Runtime is the sum of recorded latency_ms, not the complete wall-clock run duration.

Use RedactionPolicy whenever persisting reports, approval payloads, context manifests, or telemetry data. The default policy masks common fields such as API keys, tokens, passwords, secrets, authorization headers, and credentials.

Persistent Run Store​

RunReport is the durable summary for normalized events. SQLiteRunStore adds a small built-in persistence layer for task snapshots and run reports when an application wants a searchable local record without designing a database first.

from protolink import Agent, AgentCard, RunContext, SQLiteRunStore, Task

store = SQLiteRunStore("runs.db")
agent = Agent(
    AgentCard(name="worker", description="Worker", url="runtime://worker"),
    llm=llm,
    run_store=store,
)

task = Task.create_infer(prompt="produce a summary")
RunContext(run_id="run_123", session_id="session_abc").attach_to_task(task)
result = await agent.execute_task(task)

record = store.get_task_record(result.id)
recent = store.list_task_records(session_id="session_abc")

SQLiteRunStore keeps two JSON payload tables:

RecordIndexed fields
Task snapshotstask_id, state, run_id, session_id, trace_id, agent_name, timestamps
Run reportsrun_id, session_id, trace_id, agent_name, timestamp

TaskRecord and RunReportRecord​

dataclassesprotolink.TaskRecord · protolink.RunReportRecord
source
TaskRecord(
  task_id: str,
  state: str,
  run_id: str | None = None,
  session_id: str | None = None,
  trace_id: str | None = None,
  agent_name: str | None = None,
  task: dict[str, Any] = {},
  metadata: dict[str, Any] = {},
  created_at: str | None = None,
  updated_at: str = <UTC timestamp>,
)

RunReportRecord(
  run_id: str,
  session_id: str | None = None,
  trace_id: str | None = None,
  agent_name: str | None = None,
  report: dict[str, Any] = {},
  metadata: dict[str, Any] = {},
  created_at: str = <UTC timestamp>,
)

Immutable index records returned by RunStore. Payload dictionaries remain ordinary mutable values.

TaskRecord

task_idstrrequired
Stored task key.
statestrrequired
Serialized task lifecycle state.
run_idstr | Nonedefault: None
Optional run correlation copied from RunContext.
session_idstr | Nonedefault: None
Optional application-session correlation.
trace_idstr | Nonedefault: None
Optional observability trace correlation.
agent_namestr | Nonedefault: None
Storing agent.
taskdict[str, Any]default: {}
Complete serialized task snapshot.
metadatadict[str, Any]default: {}
Store-call metadata, separate from task metadata.
created_atstr | Nonedefault: None
Task creation time when available.
updated_atstrdefault: current UTC timestamp
Snapshot persistence time.

RunReportRecord

run_idstrrequired
Primary report key.
session_idstr | Nonedefault: None
Optional report-context session correlation.
trace_idstr | Nonedefault: None
Optional report-context trace correlation.
agent_namestr | Nonedefault: None
Storing agent.
reportdict[str, Any]default: {}
Complete serialized report.
metadatadict[str, Any]default: {}
Store-call metadata.
created_atstrdefault: current UTC timestamp
Persistence time.

Methods

to_dict()dict[str, Any]

Each record returns its complete raw mapping without automatic redaction.

RunStore​

protocolprotolink.RunStore
source
class RunStore(Protocol):
  def save_task(...) -> TaskRecord: ...
  def get_task(task_id: str) -> Task | None: ...
  def get_task_record(task_id: str) -> TaskRecord | None: ...
  def list_task_records(...) -> list[TaskRecord]: ...
  def save_report(...) -> RunReportRecord: ...
  def get_report(run_id: str) -> RunReport | None: ...
  def get_report_record(run_id: str) -> RunReportRecord | None: ...
  def list_report_records(...) -> list[RunReportRecord]: ...

Synchronous structural persistence contract for task snapshots and reports.

Task operations

save_task(task, *, context=None, agent_name=None, metadata=None)TaskRecord

Persists one snapshot.

get_task(task_id)Task | None
Loads the typed task.
get_task_record(task_id)TaskRecord | None
Loads the indexed record.
list_task_records(*, limit=100, session_id=None, run_id=None, state=None, agent_name=None)list[TaskRecord]

Lists newest snapshots with optional exact filters.

Report operations

save_report(report, *, run_id=None, agent_name=None, metadata=None)RunReportRecord

Persists one report.

get_report(run_id)RunReport | None
Loads the typed report.
get_report_record(run_id)RunReportRecord | None
Loads the indexed record.
list_report_records(*, limit=100, session_id=None, agent_name=None)list[RunReportRecord]

Lists newest reports with optional exact filters.

Extension contract

RunStore is a typing protocol, so application adapters implement the methods structurally. Delete operations are not part of this protocol.

SQLiteRunStore​

classprotolink.SQLiteRunStore
source
SQLiteRunStore(
  db_path: str | pathlib.Path = "runs.db",
  *,
  table_prefix: str = "protolink",
)

Dependency-free SQLite implementation using a fresh synchronous connection per operation and JSON payload columns with relational lookup indexes.

Parameters

db_pathstr | pathlib.Pathdefault: "runs.db"

SQLite database path, converted to str. Construction immediately creates tables and indexes when missing.

table_prefixstrdefault: "protolink"

Prefix for <prefix>_tasks and <prefix>_run_reports; must satisfy str.isidentifier().

Attributes

db_pathstr
Normalized database path.
table_prefixstr
Validated prefix.
tasks_tablestr
Derived task table name.
reports_tablestr
Derived report table name.

Raises

ValueError
The table prefix is not a Python identifier.
sqlite3.Error | OSError
The database cannot be opened or initialized.
Security and redaction

The store serializes raw task/report dictionaries. Apply RedactionPolicy before saving data that may contain secrets. The table prefix is validated, while record values are parameterized SQL inputs.

SQLiteRunStore.save_task​

methodSQLiteRunStore.save_task
source
store.save_task(
  task: Task,
  *,
  context: RunContext | None = None,
  agent_name: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> TaskRecord

Parameters

taskTaskrequired
Task snapshot to serialize.
contextRunContext | Nonedefault: None

Explicit correlation context; otherwise reconstructed from task metadata.

agent_namestr | Nonedefault: None
Optional storing agent.
metadatadict[str, Any] | Nonedefault: None
Separate store-record metadata.

Returns and mutation

recordTaskRecord

Stored index record. INSERT OR REPLACE overwrites an existing row with the same task ID and updates its persistence timestamp.

Raises

TypeError | ValueError
Task or metadata cannot be JSON serialized.
sqlite3.Error
Database write or commit fails.

SQLiteRunStore.list_task_records​

methodSQLiteRunStore.list_task_records
source
store.list_task_records(
  *,
  limit: int = 100,
  session_id: str | None = None,
  run_id: str | None = None,
  state: str | TaskState | None = None,
  agent_name: str | None = None,
) -> list[TaskRecord]

Parameters

limitintdefault: 100

SQL LIMIT; the implementation performs no positivity validation, and SQLite treats a negative value as no upper bound.

session_idstr | Nonedefault: None
Exact session filter.
run_idstr | Nonedefault: None
Exact run filter.
statestr | TaskState | Nonedefault: None
Exact serialized-state filter.
agent_namestr | Nonedefault: None
Exact agent filter.

Returns

recordslist[TaskRecord]
Rows ordered by updated_at DESC.

SQLiteRunStore.save_report​

methodSQLiteRunStore.save_report
source
store.save_report(
  report: RunReport,
  *,
  run_id: str | None = None,
  agent_name: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> RunReportRecord

Parameters

reportRunReportrequired
Report serialized without implicit redaction.
run_idstr | Nonedefault: None

Explicit primary key, taking precedence over report.context.run_id.

agent_namestr | Nonedefault: None
Optional storing agent.
metadatadict[str, Any] | Nonedefault: None
Separate record metadata.

Returns and mutation

recordRunReportRecord

Stored record. Context session/trace IDs are indexed when present; INSERT OR REPLACE overwrites the same run ID.

Raises

ValueError

Neither the explicit argument nor report context supplies a truthy run ID.

TypeError | sqlite3.Error

JSON serialization or database persistence fails.

SQLiteRunStore queries and deletion​

methodsSQLiteRunStore
source
store.get_task(task_id: str) -> Task | None
store.get_task_record(task_id: str) -> TaskRecord | None
store.get_report(run_id: str) -> RunReport | None
store.get_report_record(run_id: str) -> RunReportRecord | None
store.list_report_records(
  *,
  limit: int = 100,
  session_id: str | None = None,
  agent_name: str | None = None,
) -> list[RunReportRecord]
store.delete_task(task_id: str) -> None
store.delete_report(run_id: str) -> None

Parameters

task_idstrrequired

Primary key accepted by get_task(), get_task_record(), and delete_task().

run_idstrrequired

Primary key accepted by get_report(), get_report_record(), and delete_report().

limitintdefault: 100

SQL row limit for list_report_records(); the implementation does not validate positivity.

session_idstr | Nonedefault: None

Optional exact session filter for list_report_records().

agent_namestr | Nonedefault: None

Optional exact agent filter for list_report_records().

Lookup

get_task(task_id)Task | None
Restores a typed task or returns None.
get_task_record(task_id)TaskRecord | None
Returns the indexed task record or None.
get_report(run_id)RunReport | None
Restores a typed report or returns None.
get_report_record(run_id)RunReportRecord | None
Returns the indexed report record or None.

Listing and deletion

list_report_records(*, limit=100, session_id=None, agent_name=None)list[RunReportRecord]

Applies optional exact filters and returns created_at DESC. Like task listing, limit is not validated.

delete_task(task_id)None
Deletes the matching task row; a missing key is a no-op.
delete_report(run_id)None
Deletes the matching report row; a missing key is a no-op.

Raises

sqlite3.Error

A query, delete, or commit fails.

json.JSONDecodeError | model restoration error

Stored JSON is corrupt or cannot be reconstructed as a Task/RunReport.

The store is intentionally separate from the generic Storage key/value interface. Storage backs agent state such as conversations; RunStore records execution facts for lookup, replay, audit, and tests. Larger deployments can implement the same RunStore protocol against their own persistence layer.

Complete Runnable Examples​

examples/runtime_policy_and_approvals.py combines the runtime primitives in one provider-free script. It uses MockLLM to request a tool, creates a preview artifact, obtains application approval, captures normalized events, and then proves that a stricter per-run permission prevents a second side effect.

Run it from the repository root:

.venv/bin/python examples/runtime_policy_and_approvals.py

The example uses an in-memory record so it is deterministic and requires no API key, service, or network port. Its approval handler automatically approves the first action for demonstration; a real application would replace that callback with its own interactive or remote decision workflow.

examples/task_cancellation.py demonstrates live cancellation of a streamed async tool. It proves that the final side effect is not committed and prints the final normalized canceled event:

.venv/bin/python examples/task_cancellation.py

Golden Run Tests​

Golden-run tests use deterministic model/tool fixtures and assert the normalized runtime contract. They are useful for application integrations because they lock down the event and artifact sequence without depending on live model providers.

from protolink import Agent, AgentCard, InMemoryEventSink, RunContext, Task, create_llm

llm = create_llm("mock", default_response="done")
agent = Agent(
    AgentCard(name="tester", description="Golden test agent", url="runtime://tester"),
    llm=llm,
    verbosity=0,
)

task = Task.create_infer(prompt="Produce a result")
RunContext(run_id="run_golden", session_id="session_golden").attach_to_task(task)

sink = InMemoryEventSink()
async for event in agent.handle_task_streaming(task):
    await sink.emit_task_event(event, context=RunContext.from_task(task))

snapshot = [
    {
        "sequence": item["sequence"],
        "type": item["type"],
        "summary": item["summary"],
        "final": item["final"],
    }
    for item in sink.to_list()
]

Use this style for runtime compatibility tests: assert the stable event envelope, policy and approval sequence, task state, final artifacts, and context propagation. Keep volatile fields such as timestamps, UUIDs, and artifact IDs out of the golden snapshot unless the test explicitly controls them.

Relationship To Telemetry​

Runtime events and telemetry serve different layers:

  • RunEvent is for live application progress, terminal rendering, stream snapshots, and runtime assertions.
  • LocalTraceTelemetry is for replayable traces, spans, metrics, redacted payloads, and observability backends.

Both share the same run_id, trace_id, task_id, and agent metadata through RunContext, so a local UI can show live progress while telemetry records the detailed trace behind it.

As a practical rule, use events to drive what the user sees now and telemetry to investigate what happened across the complete run later.