Multi-agent work

QitOS public API: work-graph

delegate/spawn/fan_out return operation receipts; join references operation IDs. WorkGraph records work ownership and attempts; it is not a worker executor. LocalWorkScheduler resolves descriptors through application-provided callables. handoff changes the owner of the same work item and fences the old source; fork branches independently. Transfer admission is not destination task completion.

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

Signatures and fields below are extracted from the pinned source. Signatures are reference material, not standalone programs. Source links bind the same runtime baseline. Any does not imply arbitrary objects are supported; use the behavioral contract above and the linked tutorial.

Handoff persists its declaration, prepared ownership transfer and descriptor, then transfer_admitted before invoking the scheduler. The receipt acknowledges transfer admission only; outcome_unknown=True means destination execution is not yet known. A callable returning None is not task completion. The destination uses the existing restore/CAS to persist ownership_committed, then its own running and business terminal state. Late source callbacks cannot write the Session head. A restored current owner can reconcile the transfer without the source process. An admitted operation whose invocation was lost is not replayed; waiting policy, business retries and manual reconciliation remain application responsibilities.

Session

from qitos.engine.session_runtime import Session

Source @ 19f6258

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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

Usage and executable example

Usage fragment: continues with objects from the linked complete tutorial; not a standalone program.

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	''

Confirmed dispatch rejection and destination recovery

A handoff receipt with state=dispatch_not_started, admission_state=closed, and outcome_unknown=false proves the scheduler created no worker. Its terminal_receipt_ref records the safe dispatch reason. Calling handoff again with the same operation and payload returns this fact; changing the payload conflicts. Rebind the destination Agent and resources in a new composition, call composition.restore(session_id), then session.run(). No scheduler retry is needed. After an uncertain dispatch, the outcome stays unknown and recovery does not automatically replay effects. Scheduler callable return is not task completion.

Terminal reconciliation requires descriptor work/session and explicit context transfer linkage to the current owner generation; an Agent name or an ID suffix is not lineage. Historical unbound descriptors remain unresolved. Prior terminal receipts and unrelated work remain unchanged.

Compatibility: dispatch_not_started is a new member of the existing strict operation-state vocabulary. Historical readers reject this value and must upgrade; they must not reinterpret it as admitted/unknown. Historical handoff descriptors without explicit transfer linkage remain unresolved instead of attributing another work item’s terminal.