多 Agent 工作

QitOS public API: work-graph

delegate/spawn/fan_out 返回 operation receipt，join 引用 operation ID。WorkGraph 记录所有权与 attempt，本身不执行 worker。LocalWorkScheduler 通过应用提供的 callable 解析 descriptor。handoff 改变同一 work item 的 owner 并阻止旧 source；fork 独立分支。转交接收不等于目标完成；教程同时验证串行与同一 Session 的并发 handoff。

完整可运行教程 / Complete tutorial · API index

以下签名和字段由固定源码提取；签名是参考，不是可直接运行的程序。每个条目的源码链接绑定同一 runtime baseline。类型中的 Any 不代表任意对象均受支持，应结合上述行为契约和教程使用。

Handoff 依次持久化 declaration、准备好的 ownership transfer 与 descriptor， 再在调用 scheduler 前提交 transfer_admitted。该 receipt 只确认 transfer admission； outcome_unknown=True 表示尚不知道目标是否执行，callable 返回 None 不代表任务完成。 目标使用既有 restore/CAS 提交 ownership_committed，再自行记录 running 与业务终态。 迟到的源 callback 不写 Session head。恢复后的当前 owner 可仅凭持久事实 reconcile， 不依赖源进程。已 admission 但调用结果未知的 operation 不自动重放；等待策略、业务重试 与人工 reconciliation 仍由应用负责。

​

Session

from qitos.engine.session_runtime import Session

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

session = composition.session("Index notes")
result = session.run()
inspection = session.inspect()
print(session.session_id.value, result.state.final_result)

Scoped client for one durable Session identity.

The facade stores identifiers and cooperative control only. Agent state is
reconstructed from the canonical snapshot and executed by ``Engine.run``.

Session(*, engine: 'Engine[Any, Any, Any]', session_id: SessionIdentity, run_id: RunIdentity, agent_id: AgentIdentity, references: Iterable[ResolverReference], created_at: str, state_type: type[StateSchema], work_item_id: WorkItemIdentity, attempt_id: AttemptIdentity, fork_receipt: Optional[SessionForkReceipt]=None) -> None

Parameter	Type	Default
engine	'Engine[Any, Any, Any]'	required
session_id	SessionIdentity	required
run_id	RunIdentity	required
agent_id	AgentIdentity	required
references	Iterable[ResolverReference]	required
created_at	str	required
state_type	type[StateSchema]	required
work_item_id	WorkItemIdentity	required
attempt_id	AttemptIdentity	required
fork_receipt	Optional[SessionForkReceipt]	None

​

Session.delegate

delegate(agent: str, *, task: str, operation_id: str | None=None) -> Any

Parameter	Type	Default
agent	str	required
task	str	required
operation_id	str | None	None

Source

​

Session.spawn

spawn(agent: str, *, task: str, operation_id: str | None=None) -> Any

Parameter	Type	Default
agent	str	required
task	str	required
operation_id	str | None	None

Source

​

Session.fan_out

fan_out(specs: Iterable[Mapping[str, Any]], *, operation_id: str | None=None) -> Any

Parameter	Type	Default
specs	Iterable[Mapping[str, Any]]	required
operation_id	str | None	None

Source

​

Session.join

join(children: Iterable[str], *, policy: str='all', quorum: int | None=None, reducer_ref: str | None=None, reducer_digest: str | None=None, operation_id: str | None=None) -> Any

Parameter	Type	Default
children	Iterable[str]	required
policy	str	'all'
quorum	int | None	None
reducer_ref	str | None	None
reducer_digest	str | None	None
operation_id	str | None	None

Source

​

Session.handoff

handoff(agent: str, *, rationale: str='handoff', operation_id: str | None=None) -> Any

Parameter	Type	Default
agent	str	required
rationale	str	'handoff'
operation_id	str | None	None

Source

​

WorkGraph

from qitos.core.work_graph import WorkGraph

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

graph = WorkGraph.from_canonical_dict(session.inspect().work_graph)
print(len(graph.completions), len(graph.joins))

Versioned ownership graph and generation-checked record builder.

Field	Type	Default
graph_id	str	required
work_items	Dict[WorkItemIdentity, WorkItem]	field(default_factory=dict)
attempts	list[WorkAttempt]	field(default_factory=list)
edges	list[WorkEdge]	field(default_factory=list)
transfers	list[OwnershipTransfer]	field(default_factory=list)
delegations	list[DelegationRecord]	field(default_factory=list)
spawns	list[SpawnRecord]	field(default_factory=list)
fan_out_groups	list[FanOutGroup]	field(default_factory=list)
joins	list[JoinDependency]	field(default_factory=list)
cancellations	list[CancellationRequest]	field(default_factory=list)
detachments	list[DetachmentRecord]	field(default_factory=list)
completions	list[WorkCompletion]	field(default_factory=list)
late_results	list[LateResult]	field(default_factory=list)
budget_allocations	list[BudgetAllocation]	field(default_factory=list)
capability_allocations	list[CapabilityAllocation]	field(default_factory=list)
operation_receipts	list[WorkOperationReceipt]	field(default_factory=list)
schema_version	str	WORK_GRAPH_SCHEMA_VERSION

​

WorkGraph.from_canonical_dict

from_canonical_dict(payload: Mapping[str, Any]) -> 'WorkGraph'

Parameter	Type	Default
payload	Mapping[str, Any]	required

Source

​

WorkItem

from qitos.core.work_graph import WorkItem

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

work = graph.work_items[session.work_item_id]
print(work.owner)

Field	Type	Default
work_item_id	WorkItemIdentity	required
session_ref	SessionIdentity	required
task_ref	str	required
lifecycle	WorkLifecycle	required
owner	WorkOwner	required
parent_work_item_id	WorkItemIdentity | None	None
detached	bool	False
budget_allocation_ref	str | None	None
capability_allocation_ref	str | None	None
context_transfer_ref	str | None	None

​

WorkAttempt

from qitos.core.work_graph import WorkAttempt

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

from dataclasses import fields
print([field.name for field in fields(WorkAttempt)])

Field	Type	Default
attempt_id	AttemptIdentity	required
work_item_id	WorkItemIdentity	required
owner_generation	int	required
state	AttemptState	required
worker_ref	str | None	None

​

DurableWorkRuntime

from qitos.engine.work_runtime import DurableWorkRuntime

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

runtime = DurableWorkRuntime(LocalWorkScheduler(Resolver(), max_workers=2))
composition.runtime.work_runtime = runtime

Idempotent declaration/dispatch protocol over one canonical WorkGraph.

DurableWorkRuntime(scheduler: WorkScheduler, *, policy: WorkRuntimePolicy | None=None) -> None

Parameter	Type	Default
scheduler	WorkScheduler	required
policy	WorkRuntimePolicy | None	None

​

LocalWorkScheduler

from qitos.engine.work_runtime import LocalWorkScheduler

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

scheduler = LocalWorkScheduler(Resolver(), max_workers=2)
# Resolver.resolve(descriptor) returns a bounded callable; see the full lesson.

Bounded local reference scheduler; futures are never persisted.

LocalWorkScheduler(resolver: WorkResolver, *, max_workers: int=4, queue_capacity: int=64) -> None

Parameter	Type	Default
resolver	WorkResolver	required
max_workers	int	4
queue_capacity	int	64

​

WorkRuntimeError

from qitos.engine.work_runtime import WorkRuntimeError

Source @ 19f6258

用法与可执行示例

用法片段：接续上方完整教程中的对象，不是独立程序。

try:
    source.spawn("notes_agent", task="Attempt after handoff")
except WorkRuntimeError as error:
    print(error.code)

Typed scheduler/admission/idempotency failure.

WorkRuntimeError(code: str, message: str, *, operation_id: str='') -> None

Parameter	Type	Default
code	str	required
message	str	required
operation_id	str	''

​

确认未调度与目标恢复

Handoff receipt 的 state=dispatch_not_started、admission_state=closed、 outcome_unknown=false 表示 scheduler 确认没有创建 worker；terminal_receipt_ref 记录安全的 dispatch 原因。同 operation 和 payload 重试返回该事实，修改 payload 会冲突。 在新 composition 重新绑定目标 Agent 与资源后调用 composition.restore(session_id)， 再调用 session.run()；无需再次调度。dispatch 后结果不明仍保持 unknown，不自动重放 未知 effect。Scheduler callable 返回不代表业务完成。

终态关联需要 descriptor work/session、显式 context transfer 引用与当前 owner generation； 不根据 Agent 名称或 ID 后缀推断。历史缺少绑定的 descriptor 保持未决；既有终态与无关 work 不变。

兼容说明：dispatch_not_started 是既有严格 operation-state 闭集中的新增值。旧 reader 会拒绝该值，需要升级，不能将其解释为 admitted/unknown。缺少显式 transfer 关联的历史 handoff descriptor 保持未关联，不能借用另一个 work item 的终态。