Forking API

Live Run Forking lets an agent explore multiple solution branches in parallel and merge the best result. Enable it via forking=True on create_deep_agent. See Live Run Forking for the conceptual overview.

LiveForkCapability

pydantic_deep.capabilities.forking.LiveForkCapability dataclass

Bases: AbstractCapability[Any]

Capability that wires Live Run Forking into an agent.

Parameters:

Name	Type	Description	Default
max_branches	int	Maximum branches per fork.	10
max_depth	int	Maximum fork nesting depth - 2 allows one level of fork-of-fork.	2
store	ForkStateStore | None	Optional :class:ForkStateStore. Defaults to :class:InMemoryForkStateStore.	None
test_command	str | None	Optional shell command run against each branch's materialised tree during :meth:ForkCoordinator.resolve to feed the test_pass_ratio confidence signal. None leaves the ratio at None for every branch - :func:compute_confidence then keeps its cap-at-0.65 safety rail active, identical to "no test signal". Only honoured when the parent backend is a :class:~pydantic_ai_backends.LocalBackend.	None
test_timeout_s	float	Wall-clock cap (seconds) per branch test run. On timeout the branch's test_pass_ratio is None (treated as "no signal"), not 0.0. Independent of branch_budget_usd - runner time is not LLM cost.	60.0

The owning agent reference is set by create_deep_agent() after the Agent is constructed (mirrors how agent._task_manager is set today).

latest_messages property

Snapshot of the parent run's most recent message list.

Returns a copy so callers can't mutate the capability's internal state. Updated on every before_model_request; used by fork_run to seed each branch's history at the moment of the fork call.

for_run(ctx) async

Return a fresh per-run capability with an independent coordinator.

Preserves an unresolved coordinator from a previous turn rather than overwriting it. If the previous parent run forked but did not call merge_or_select before yielding back to the user, the coordinator stays on deps.fork_coordinator so the next turn (or the CLI adopter) can still resolve or abort it; the new capability clone takes ownership via the capability back-reference so per-run state (e.g. latest_messages) flows through the right instance.

after_run(ctx, *, result) async

Anchor for the post-turn stash protocol - currently a no-op.

The coordinator survives a parent turn ending because :meth:for_run refuses to overwrite an unresolved one on the next turn (see above). This hook exists as a documented anchor so future strategy changes (eager artefact cleanup on abort, post-turn telemetry, partial-history persistence) plug in here without restructuring the lifecycle. result is returned unchanged.

before_model_request(ctx, request_context) async

Snapshot the latest message list.

For parent runs the snapshot lands on :attr:_latest_messages so :meth:ForkCoordinator.fork can seed each branch's history.

For branch runs (identified by ctx.deps._branch_id being non-None - set by :meth:ForkCoordinator.fork) the snapshot is forwarded to the parent coordinator via :meth:ForkCoordinator.capture_partial_history, so that if a budget watcher cancels the branch the merge resolver still has a history to return when the branch is picked as winner.

ForkCoordinator

pydantic_deep.toolsets.forking.ForkCoordinator

Owns per-branch state for one parent run.

A fresh coordinator is allocated by :meth:LiveForkCapability.for_run, so concurrent parent runs of the same agent never share state.

Parameters:

Name	Type	Description	Default
agent	Any	The owning agent - used to spawn branch agent.run() tasks.	required
parent_deps	DeepAgentDeps	The parent run's deps; cloned per branch via :func:clone_for_branch.	required
max_branches	int	Maximum number of branches per fork.	required
max_depth	int	Maximum fork nesting depth.	required
store	ForkStateStore	The :class:ForkStateStore used to persist :class:ForkHandle.	required
checkpoint_store	CheckpointStore | None	Optional explicit checkpoint store. When None, the coordinator falls back to parent_deps.checkpoint_store.	None
test_command	str | None	Optional shell command run against each branch's materialised tree during :meth:resolve to feed the test_pass_ratio confidence signal. None disables the runner - the cap-at-0.65 safety rail then keeps auto_with_fallback falling through to the manual picker. Restricted to :class:~pydantic_ai_backends.LocalBackend parents; non-local parents always produce None. SECURITY: this is operator-configured and runs with the parent process's full os.environ inherited (only UV_NO_SYNC=1 is added). The environment is deliberately NOT filtered, since most test commands need PATH / HOME / language runtime vars to function. Operators must therefore avoid configuring a test_command that could exfiltrate secrets held in the environment (e.g. API keys), as those vars are visible to it.	None
test_timeout_s	float	Wall-clock cap (seconds) per branch test run. On timeout the branch's test_pass_ratio is None (treated as "no signal"), not 0.0.	60.0

fork_id property

The active fork's id, or None if fork() has not been called yet.

Exposed so consumers (e.g. the diff_branches tool) can validate caller-supplied fork_id without reaching into the coordinator's private _handle attribute.

handle property

The :class:ForkHandle returned by :meth:fork, or None before fork.

Read-only public accessor for the same value :meth:fork returns. Cross-package consumers (the CLI adopter, debug inspectors) read this instead of reaching into _handle.

is_resolved property

True when the coordinator no longer owns live branch state.

Resolved iff either the coordinator has not yet forked (_handle is None) or every branch's overlay has been released (rt.overlay is None) - which only happens inside :meth:merge_or_select (winner flushed, losers cancelled) and :meth:aclose (abort). This is the canonical "safe to discard" signal used by the CLI adopter and :meth:LiveForkCapability.for_run to distinguish a fork that needs preserving from one that does not.

fork(specs, *, parent_history, isolation=None, strategy=None, aggregate_budget_usd=None) async

Spawn len(specs) branch tasks and return a handle.

Parameters:

Name	Type	Description	Default
specs	list[BranchSpec]	Branch definitions; len(specs) must not exceed :attr:max_branches.	required
parent_history	list[Any]	Parent run's message snapshot at fork time.	required
isolation	BranchIsolation | None	Per-branch isolation overrides (defaults to :class:BranchIsolation).	None
strategy	MergeStrategy | None	Merge strategy (currently kind="manual" only).	None
aggregate_budget_usd	float | None	Optional fork-wide budget cap; when set, hitting it terminates every still-running branch with state="aggregate_budget_exhausted". Overrides the value passed to :meth:__init__. Enforcement is best-effort under concurrent callbacks - see the "Aggregate budget enforcement" note in the live-fork doc.	None

Raises:

Type	Description
ValueError	If specs is empty.
ForkBranchLimitError	If len(specs) > max_branches.
ForkDepthLimitError	If parent's _fork_depth >= max_depth.

run_on_branch(branch_id, user_message) async

Start a new turn on a finished branch with user_message.

The branch must be in done state. Seeds the new turn's message history from the previous run's all_messages() (which already holds the full conversation), spawns a new asyncio.Task, replaces runtime.task, and re-attaches the done-callback so the status transitions back through running → done.

Returns the spawned task so the caller can await it if needed.

Raises:

Type	Description
ValueError	If branch_id is unknown.
RuntimeError	If the branch is still running.

terminate_branch(branch_id, *, reason=None) async

Cancel a branch task and mark its terminal status.

Parameters:

Name	Type	Description	Default
branch_id	str	Branch to cancel.	required
reason	str | None	When "budget_exhausted" or "aggregate_budget_exhausted", the branch's status is set to the matching state and its error field is populated with a debug-friendly message. Otherwise the terminal state is "terminated".	None

Only sets a new terminal state when the task is still running; if it already completed (success / failure / earlier cancellation), _on_done has already written the correct terminal state and we must not overwrite it. Idempotent: a second call for the same branch_id (e.g. when the aggregate watcher races with the per-branch watcher) is a no-op.

iter_pending_approvals()

Return (branch_id, request) for every branch currently suspended on approval.

The TUI poll loop uses this instead of reading :attr:BranchRuntime.pending_approval directly, so the coordinator owns the contract about who may inspect that field.

merge_or_select(action) async

Resolve the fork by picking a winner.

action="pick:<branch_id>" awaits the winning branch's task, cancels and discards the others, replays the winner's overlay onto the parent backend, releases every overlay, and saves a post-fork:<fork_id> checkpoint when checkpointing is available.

abort_fork() async

Discard the entire fork without merging - releases overlays, cancels tasks.

Use when every branch has failed (or otherwise become unmergeable) so the fork can be resolved without picking a winner. Mirrors the cleanup half of :meth:merge_or_select but without flushing any overlay onto the parent backend.

Returns the list of branch ids that were aborted. After this call, :attr:is_resolved becomes True and a new fork can be started on the same coordinator.

resolve(strategy=None) async

Dispatch on :attr:MergeStrategy.kind - judge runs for non-manual modes.

• "manual" → early-return ResolveOutcome(committed=False, auto_eligible=False, verdict=None, ...); the caller picks via :meth:merge_or_select.
• "auto" → judge picks, merge_or_select fires immediately.
• "auto_with_fallback" → judge picks. If the combined confidence is at or above :attr:MergeStrategy.confidence_threshold the commit is deferred to the caller (committed=False, auto_eligible=True) so the CLI's acceptance widget can offer an override; otherwise auto_eligible=False and the caller opens the manual picker preselected.
• "vote" → multiple judges evaluate concurrently; majority wins, ties broken by highest individual confidence; merge_or_select fires immediately on the synthetic majority verdict.

The judge's result.usage() rides on :attr:ResolveOutcome.judge_usage (summed across judges for "vote") so the caller can attribute cost without faking a cost_category field on pydantic-ai-shields' CostTracking.

inspect_branches()

Return a snapshot of every branch's current status.

fork_cost()

Return per-branch and aggregate cost for the active fork.

Returns:

Type	Description
ForkCostSummary	class:ForkCostSummary with one :class:BranchCost entry per
ForkCostSummary	branch. aggregate_usd sums the cumulative_usd values of
ForkCostSummary	branches whose cost is known (skips branches whose model has no
ForkCostSummary	pricing or has not produced a tracked run yet); when no branch
ForkCostSummary	has a known cost, aggregate_usd is None.

Raises:

Type	Description
RuntimeError	If called before :meth:fork has been invoked.

capture_partial_history(branch_id, messages)

Record a branch's latest message snapshot for merge fallback.

Called by :class:LiveForkCapability.before_model_request on every branch run so that, if the branch is later cancelled by a budget watcher, :meth:merge_or_select has a snapshot to return when the user picks the exhausted branch as winner. Silently ignored when the branch is unknown - defensive against late callbacks after aclose().

aclose() async

Cancel every outstanding branch task - used on parent cancellation.

Also runs materializer.cleanup() so the on-disk fork directory is removed on abort (unless keep_artifacts is set), mirroring the merge-resolution cleanup. Safe to call multiple times.

create_fork_toolset

pydantic_deep.toolsets.forking.create_fork_toolset(id='deep-forking')

Build the four-tool forking toolset wired to the per-run coordinator.

ForkStateStore

pydantic_deep.toolsets.forking.ForkStateStore

Bases: Protocol

Protocol for fork state storage backends.

Stores ForkHandle records keyed by fork_id. Forks live for the duration of the parent run; on process restart, in-memory state is lost.

save(handle) async

Save or overwrite a fork handle.

get(fork_id) async

Return the handle for fork_id or None if unknown.

list_all() async

Return all stored fork handles.

remove(fork_id) async

Remove a handle by id. Returns True if it existed.

InMemoryForkStateStore

pydantic_deep.toolsets.forking.InMemoryForkStateStore

Default in-memory fork state store.

build_diff_report

pydantic_deep.toolsets.forking.build_diff_report(fork_id, runtimes, *, paths_filter=None)

Build a :class:BranchDiffReport from a list of branch runtimes.

This is the public entry point for the diff explorer. The agent-facing :func:diff_branches tool calls into this; downstream consumers (CLI merge picker, judge) call it directly for programmatic access.

Parameters:

Name	Type	Description	Default
fork_id	str	Identifier of the fork being inspected - echoed in the report's fork_id field.	required
runtimes	list[BranchRuntime]	Branch runtimes to compare; usually list(coordinator.branches.values()).	required
paths_filter	list[str] | None	Optional path list; when provided, only these paths appear in the report. Untouched filtered paths surface as agreement="unanimous_no_change" for transparency.	None

Returns:

Name	Type	Description
A	BranchDiffReport	class:BranchDiffReport covering every touched path (or every
	BranchDiffReport	filtered path). See module docstring for read-consistency caveat.

JudgeAgent

pydantic_deep.toolsets.forking.JudgeAgent

Thin :class:Agent wrapper that picks the winning branch via structured output.

Holds an internal pydantic-ai :class:Agent with :class:JudgeVerdict as output_type. The agent is built once at construction; concurrent evaluate calls reuse the same underlying agent. The system prompt is the module-level :data:JUDGE_SYSTEM_PROMPT; per-call context is composed by :func:_build_judge_prompt.

evaluate(goal, diff_report, outcomes) async

Run the judge and return (verdict, usage).

usage is the result.usage property value (pydantic-ai exposes it as a property, not a method) - the coordinator bubbles it up via :attr:ResolveOutcome.judge_usage for cost attribution. The judge never receives full per-branch message history, only the goal + diff + outcome summaries.

compute_confidence

pydantic_deep.toolsets.forking.compute_confidence(signals, judge_confidence)

Combine the three signals with the judge's self-reported confidence.

Heuristic = 0.4 * quality_spread + 0.4 * test_pass_ratio + 0.2 * internal_consistency. If signals.test_pass_ratio is None the test slot is treated as 0.0 for the weighted sum AND the heuristic is capped at 0.65 before multiplying by judge_confidence - the safety rail that keeps auto_with_fallback defaulting to manual when the test signal is missing. The product is clamped to [0.0, 1.0].

Coordinator

pydantic_deep.toolsets.forking.coordinator.ForkBranchLimitError

Bases: Exception

Raised when a fork_run call exceeds max_branches.

pydantic_deep.toolsets.forking.coordinator.ForkDepthLimitError

Bases: Exception

Raised when a fork_run call from within a branch exceeds max_depth.

Isolation

pydantic_deep.toolsets.forking.isolation.BranchOverlay

Copy-on-write backend overlay for a single branch.

Reads consult the overlay first; if a path has not been written in this branch, the read falls through to the parent backend. Writes go to the overlay only and are logged to _changes for downstream consumers (diff builder, materializer, judge).

The overlay implements the subset of :class:BackendProtocol exercised by branch operations (read, write, edit, ls_info, glob_info, grep_raw, read_bytes). Forwarding for the latter three merges overlay and parent results with the overlay taking precedence.

changes()

Return the temporal-ordered list of writes recorded in this overlay.

deleted()

Paths the branch has explicitly removed via :meth:delete.

Returns a copy so callers can't mutate the overlay's internal tombstone set. Mirrors :meth:changes - same convention.

snapshot(parent_root, *, include_venv=False)

Yield a tempdir presenting an isolated, branch-flavoured view of parent_root.

Parent files are detached file-level copies (the subprocess reads and may rewrite them in place without ever touching the real parent file - copying is the core isolation guarantee, see :func:_copy_tree); overlay writes are materialised as real files; deletions remove the corresponding entry. The directory is cleaned up on context exit regardless of how the body returns.

parent_root is taken explicitly rather than read off :attr:_parent because not every backend has a root_dir (StateBackend does not); the caller decides whether snapshotting is meaningful before invoking this.

When include_venv=True and parent_root / ".venv" exists, a symlink to it is added to the snapshot. .venv is normally in :data:_SNAP_SKIP_DIRS to keep snapshot creation lean, but a test runner (pytest, uv run, etc.) typically needs the virtual environment on PATH - opting in restores it without copying. Off by default so the existing execute consumer keeps the slim layout.

delete(path)

Mark path deleted in this branch - propagated on merge.

After this returns, exists / read / read_bytes for path behave as if the file is gone, even when path lives in the parent backend. A FileChange(op="delete") is appended to :attr:_changes so :meth:flush_to can replay the deletion onto the parent. The parent's bytes are snapshotted on first touch so third-actor-delete conflict detection works - subject to the first-touch (not fork-time) limitation documented in :meth:_snapshot_parent_on_first_touch.

Writing the same path afterwards "un-deletes" it (see :meth:write).

record_mkdir(path)

Record a directory creation detected by _propagate_mutations.

record_rmdir(path)

Record a directory deletion detected by _propagate_mutations.

exists(path)

Public BackendProtocol predicate - overlay first, fall through to parent.

A branch "sees" any file present in either layer, mirroring the copy-on-write read semantics: written-by-this-branch files take precedence, otherwise the parent backend answers. Paths the branch has deleted via :meth:delete - or that live inside a deleted directory - are hidden regardless of parent presence.

attach_materializer(materializer, branch_label)

Wire a :class:ForkMaterializer into this overlay.

After this call every successful write / edit is mirrored to disk under the materializer's branches/{branch_label}/ subtree, and the parent backend's bytes for each touched path are captured lazily on first touch via :meth:ForkMaterializer.snapshot_parent_path. Note this is a first-touch snapshot, not a fork-time one - see the limitation in :meth:_snapshot_parent_on_first_touch for the conflict-detection gap when a third actor writes a path before this branch touches it.

flush_to(parent, pre_flush_snapshot=None)

Replay this overlay's writes onto parent.

Parameters:

Name	Type	Description	Default
parent	BackendProtocol	Destination backend. Usually the parent run's backend; for fork-of-fork it is the OUTER branch's overlay (which is itself a :class:BranchOverlay) - propagation up one level works without special casing.	required
pre_flush_snapshot	dict[str, bytes | None] | None	Optional mapping of path → parent bytes snapshotted when this branch first touched the path (or None for "did not exist"). When supplied, flush_to compares each touched path's current parent bytes against the snapshot and records a conflict for divergent paths. Both modified-by-third-actor and deleted-by-third-actor cases land in conflicts - except the case where the third actor wrote the path between the fork and this branch's first touch, since the snapshot already captured the third actor's bytes (see the limitation in :meth:_snapshot_parent_on_first_touch).	None

Returns:

Name	Type	Description
A	FlushReport	class:FlushReport with applied_paths (one entry per
	FlushReport	successfully-replayed path, last-write-wins), applied_changes
	FlushReport	(every replayed op - ≥ len(applied_paths)), conflicts
	FlushReport	(divergent paths - these are NOT replayed so the newer parent
	FlushReport	content is preserved), and errors (per-write failures -
	FlushReport	flush_to never aborts on the first failure).

Order: writes are replayed in :attr:_changes order (temporal), so a sequence write A → edit A → write B results in parent reflecting the final overlay state for both A and B.

pydantic_deep.toolsets.forking.isolation.clone_for_branch(deps, isolation)

Clone DeepAgentDeps for a branch according to isolation.

See :class:BranchIsolation for per-flag semantics. Memory isolation follows the backend (memory lives at {memory_dir}/{agent_name}/MEMORY.md inside the backend); the memory flag is recorded for forward-compat but has no separate effect here. team_bus is a no-op when the teams capability is not enabled on the parent run; when enabled it propagates the parent bus reference by default.

Editor

pydantic_deep.toolsets.forking.editor.EditorDetector

Static utility - detection and invocation are stateless.

detect(env=None) staticmethod

Return the kind of editor to use based on env + PATH.

Parameters:

Name	Type	Description	Default
env	Mapping[str, str] | None	Optional mapping to read the override variable from; when None reads os.environ directly. Tests pass an explicit mapping so they don't have to manipulate the process env.	None

invoke(kind, parent, branch_paths, *, custom_cmd=None) staticmethod

Launch the detected editor against parent and branch_paths.

Returns:

Type	Description
list[Popen[bytes]]	The list of spawned :class:subprocess.Popen handles - empty
list[Popen[bytes]]	for the TUI fallback (caller opens the in-TUI explorer
list[Popen[bytes]]	instead). For PyCharm the list always has length 1; for VS
list[Popen[bytes]]	Code it has length len(branch_paths) (one Popen per
list[Popen[bytes]]	branch, each diffing parent vs that one branch - VS Code only
list[Popen[bytes]]	does 2-way diff).

Parameters:

Name	Type	Description	Default
kind	EditorKind	Editor kind from :meth:detect.	required
parent	Path | None	Path to the materialised parent snapshot, or None to open a branch-only diff (branches compared directly, no parent file passed to the editor).	required
branch_paths	list[Path]	One path per branch (materialised under branches/{label}/{path}).	required
custom_cmd	str | None	Command template used when kind == "custom". Tokens {parent} and {branches} are substituted with the parent path (empty string when parent is None) and a space-separated list of branch paths respectively. When None and the kind is "custom", the environment variable PYDANTIC_DEEP_DIFFTOOL is read at invoke time.	None

Judge

pydantic_deep.toolsets.forking.judge.count_retry_parts(messages)

Count every RetryPromptPart in messages (any source - stuck-loop or other).

pydantic_deep.toolsets.forking.judge.count_stuck_loop_hits(messages)

Count RetryPromptPart parts in messages that match a stuck-loop marker.

Best-effort heuristic - relies on the marker strings in :data:_STUCK_LOOP_MARKERS matching the ModelRetry messages raised by :class:StuckLoopDetection. If the stuck-loop module's message wording changes, update the markers list.

Materializer

pydantic_deep.toolsets.forking.materializer.ForkMaterializer dataclass

Real-time disk mirror for one fork's overlays.

Parameters:

Name	Type	Description	Default
root	Path	Base directory - typically .pydantic-deep/forks/{fork_id}.	required
fork_id	str	The fork identifier; surfaced in the manifest.	required
keep_artifacts	bool	When True, :meth:cleanup is a no-op so the disk layout survives the merge for post-hoc inspection.	False

The :attr:root directory is created on construction along with its parent/ and branches/ subdirectories. The manifest is written eagerly with the initial branch list (whatever the caller passes to :meth:update_manifest after construction) so the layout is discoverable even before any writes happen.

parent_path(path)

Return on-disk location of the parent snapshot for path.

branch_path(branch_label, path)

Return on-disk location of a branch's materialised path.

snapshot_parent_path(path, content)

Capture the parent backend's content for path at first touch.

Called lazily on the first overlay write for a path so we don't eagerly walk the parent backend at fork time (we don't know which paths will be touched until the branch agent writes them). content is None when the path didn't exist in the parent at first touch - recorded as a sentinel so the deletion-by-third-actor conflict path is detectable later.

Caveat: because the capture happens at first touch rather than at fork time, a path written by a third actor between the fork and this branch's first touch is snapshotted with the third actor's bytes, so :meth:BranchOverlay.flush_to cannot flag it as a conflict. See :meth:~pydantic_deep.toolsets.forking.isolation.BranchOverlay._snapshot_parent_on_first_touch.

pre_flush_snapshot()

Return the pre-fork parent snapshot consumed by flush_to.

flush_change(branch_label, change, content)

Mirror one overlay change to disk under branches/{label}/.

flush_delete(branch_label, change)

Remove the on-disk mirror for a deleted path.

Mirrors the branch's end-state so external diff tools comparing parent/<path> against branches/{label}/<path> see "file removed in branch". A no-op when the path was never materialised in this branch.

update_manifest(statuses)

Refresh manifest.json with the latest per-branch status.

cleanup()

Remove the fork directory unless keep_artifacts is set.

Forking Types

pydantic_deep.types.BranchSpec dataclass

Specification for a single branch in a fork.

Attributes:

Name	Type	Description
label	str	Human-readable branch label (e.g. "approach_a").
steer	str	First user message delivered to the branch agent - the instruction that differentiates this branch from siblings.
model	str | None	Optional model override for the branch. None inherits the parent's model.
budget_usd	float | None	Optional per-branch USD budget. Enforced by :class:_BudgetWatcher: when the branch's CostTracking cumulative cost crosses this cap the branch is cancelled and transitions to :data:BranchState "budget_exhausted".
extra_instructions	str | None	Optional extra instructions appended to the branch's system prompt.

pydantic_deep.types.BranchState = Literal['running', 'done', 'failed', 'terminated', 'budget_exhausted', 'aggregate_budget_exhausted'] module-attribute

Lifecycle states of a single branch.

budget_exhausted means the per-branch cap was crossed: the watcher cancels the branch mid-run and the partial-history snapshot becomes the durable record. aggregate_budget_exhausted means the fork-wide cap was hit and may interrupt any still-running branch mid-stream.

pydantic_deep.types.BranchStatus dataclass

Runtime status snapshot of a single branch.

pydantic_deep.types.BranchOutcome dataclass

Per-branch summary the judge sees in its prompt.

Intentionally narrow - no full message history. The judge sees the original goal, the structured diff report, and one BranchOutcome per branch. Keeps the prompt bounded and the cost predictable.

error_count is typed as int but in practice is 0 or 1 per branch (derived from the terminal branch state). Richer tool-level error counting is a forward-compatible extension.

pydantic_deep.types.BranchIsolation dataclass

Isolation flags controlling what state branches share with the parent.

Defaults match the project's per-branch isolation policy: history always copies; backend, memory, todos copy by default; message_queue is isolated; team_bus is shared (peer-to-peer bus - branches can talk to each other by default).

Only backend="copy" and message_queue="isolated" are exercised by the current fork pipeline; the other share / share_readonly values are accepted for forward compatibility.

pydantic_deep.types.BranchChange dataclass

One branch's outcome for a single path within a BranchDiffReport.

State-level aggregate: describes the END STATE of a path on a single branch relative to the parent backend, classified into one :data:BranchDiffOperation ("created" / "modified" / "deleted" / "untouched"). The classification is derived from parent existence + overlay content, NOT from :class:FileChange.op - a branch that issues an op="write" on a path absent from the parent yields operation="created"; the same op="write" on a path present in the parent yields operation="modified".

Not to be confused with :class:FileChange, which is the event-level log of individual writes/edits/deletes that produced this state.

pydantic_deep.types.BranchCost dataclass

Per-branch cost snapshot - element of :class:ForkCostSummary.

cumulative_usd is the externally-facing name for the upstream CostTracking.total_cost value; the rename happens at the :meth:ForkCoordinator.fork_cost boundary. None indicates pricing was unavailable for the branch's model (e.g. an unrecognised model in the genai-prices catalogue) - the branch still runs, but its budget cap is effectively disabled.

pydantic_deep.types.BranchDiffAgreement = Literal['unanimous_change', 'unanimous_no_change', 'split', 'unique'] module-attribute

Cross-branch classification of a single path's outcomes.

• unanimous_change: ≥2 branches touched and all touchers produced identical content.
• unanimous_no_change: no branch touched the path (only surfaces when an explicit paths filter pulls it into the report for transparency).
• split: ≥2 branches touched and their outcomes differ.
• unique: exactly one branch touched (others left the path alone).

pydantic_deep.types.BranchDiffOperation = Literal['created', 'modified', 'deleted', 'untouched'] module-attribute

What a single branch did to a given path, relative to the parent backend.

"deleted" surfaces when a branch removed the path — either via the delete_file agent tool, or via a shell rm invoked through execute against a :class:~pydantic_ai_backends.LocalBackend parent (the snapshot mutation tracker propagates the deletion back into the overlay). The classifier _classify_agreement treats deletions like any other operation: all-deleters → unanimous_change, mixed → split, single deleter → unique.

pydantic_deep.types.BranchDiffReport dataclass

Typed cross-branch diff returned by :func:diff_branches.

Bundles a per-path :class:PathDiff list with a :class:DiffSummary of aggregate metrics (agreement score, unanimous vs split path counts, per-branch unique-touch counts) so callers can render or score the fork's divergence without re-walking individual overlays.

pydantic_deep.types.ConfidenceSignals dataclass

Three weighted signals combined by compute_confidence.

• quality_spread - 1 - agreement_score from :class:BranchDiffReport. High = branches diverged meaningfully; weight 0.4.
• test_pass_ratio - passed / total tests in the winner branch. None when no per-branch test signal is available; the cap-at-0.65 safety rail kicks in. Weight 0.4.
• internal_consistency - 1 - (retries + stuck_loop_hits) / turns for the winner; clamped to [0.0, 1.0]. Weight 0.2.

The pipeline currently ships without a per-branch test-runner hook, so test_pass_ratio is None in practice; the safety rail caps the combined heuristic at 0.65 in that case and auto_with_fallback falls through to manual resolution until a test-signal hook lands (see follow-ups in the live-fork doc).

pydantic_deep.types.DiffSummary dataclass

Aggregate metrics for a BranchDiffReport.

agreement_score is 1.0 - split_paths / max(total_paths_touched, 1). Only contested paths - those touched by more than one branch with diverging end states (agreement == "split") - count against the score. Paths touched by exactly one branch (agreement == "unique") count toward neither agreement nor disagreement, so maximal non-overlapping divergence (every branch editing different files) yields agreement_score == 1.0. This is intentional: the metric measures same-path conflict, not breadth of divergence. per_branch_unique captures the orthogonal "how much did each branch touch alone" signal. The judge consumes 1 - agreement_score as quality_spread.

pydantic_deep.types.FileChange dataclass

Single overlay mutation event recorded by :class:BranchOverlay.

Event-level log entry: one record per successful write, edit, or delete on the branch overlay. The temporal-ordered list returned by :meth:BranchOverlay.changes is the data spine consumed by every downstream consumer of the fork pipeline:

• :func:build_diff_report - uses path to know which files a branch touched.
• :class:ForkMaterializer - uses op to replay write / edit / delete semantics on the on-disk mirror.
• :class:JudgeAgent - uses timestamp for temporal heuristics when scoring branch outcomes.

Not to be confused with :class:BranchChange, which is a state-level aggregate describing a branch's per-path outcome relative to the parent backend ("created" / "modified" / "deleted" / "untouched"). FileChange logs individual operations; BranchChange summarises their cumulative effect.

pydantic_deep.types.FlushError dataclass

One per-write failure observed by :meth:BranchOverlay.flush_to.

flush_to never aborts on the first failure - it accumulates errors for the caller to surface alongside the changes that did land. Surfaced on :attr:MergeResult.errors so the CLI / agent can report partial success without losing track of what didn't apply.

pydantic_deep.types.FlushReport dataclass

Outcome of replaying a :class:BranchOverlay onto the parent backend.

Produced by :meth:BranchOverlay.flush_to during merge_or_select when the user picks a winner with default-flush semantics. The fields flow through to :class:MergeResult so the CLI / agent can render "Merged: kept branch X · N files applied · conflicts: … · errors: N" style notifications.

• applied_paths lists paths whose final overlay content was written; a path's last write/edit wins, multiple in-overlay edits to the same path collapse to one entry.
• applied_changes counts every replayed op (≥ len(applied_paths)).
• conflicts lists paths where the parent's pre-flush content diverged from the pre-fork snapshot - both modified-by-third-actor and deleted-by-third-actor cases land here. Conflicting paths are NOT replayed onto the parent (non-destructive): the newer parent content is preserved and the path is excluded from applied_paths so the caller can resolve the conflict manually.
• errors is one :class:FlushError per per-write failure (e.g. parent WriteResult.error non-empty or parent raised). The failing path is excluded from applied_paths; remaining writes still flush.
• deleted_paths lists paths the branch removed via the delete agent tool that were successfully propagated to the parent backend on merge. Paths whose deletion failed (parent raised, or the parent backend cannot delete) land in errors instead.

pydantic_deep.types.ForkCostSummary dataclass

Output of the fork_cost(fork_id) tool.

Sums :attr:BranchCost.cumulative_usd across branches with a non-None cost; branches with None are omitted from the aggregate to avoid misleading partial sums.

pydantic_deep.types.ForkHandle dataclass

Handle returned by ForkCoordinator.fork() identifying a live fork.

pydantic_deep.types.JudgeVerdict

Bases: BaseModel

Structured output of :meth:JudgeAgent.evaluate.

Pydantic BaseModel (not dataclass) because pydantic-ai's output_type contract requires a BaseModel-shaped schema for structured output.

pydantic_deep.types.MergeResult dataclass

Result returned by ForkCoordinator.merge_or_select().

pydantic_deep.types.MergeStrategy dataclass

How a fork is resolved into a single winning branch.

:attr:kind selects the resolution mode:

• "manual" - caller picks via merge_or_select(action="pick:<id>").
• "auto" - :class:JudgeAgent picks; coordinator commits immediately.
• "auto_with_fallback" - judge picks; if effective confidence is at or above :attr:confidence_threshold the commit is deferred to the caller (so the acceptance widget can offer an override), otherwise the caller opens the manual picker with the judge's pick preselected. This is the default.
• "vote" - multiple judges (default: Haiku + GPT-mini + Gemini Flash) evaluate independently; majority wins, tie broken by highest confidence; coordinator commits immediately.

pydantic_deep.types.PathDiff dataclass

Per-path slice of a BranchDiffReport: parent state + every branch's outcome.

pydantic_deep.types.ResolveOutcome dataclass

Outcome envelope returned by :meth:ForkCoordinator.resolve.

Three commit semantics live behind the same envelope so the caller (CLI) can branch cleanly:

• committed=True: the coordinator already ran merge_or_select; see :attr:merge_result. Modes that hit this path: "auto" and "vote".
• committed=False, auto_eligible=True: above-threshold "auto_with_fallback". Commit is deferred to the caller so the acceptance widget can offer an [o] override before the merge fires.
• committed=False, auto_eligible=False: below-threshold "auto_with_fallback" (caller opens picker preselected) OR "manual" (no judge ran, caller picks).

:attr:judge_usage carries the judge's result.usage (summed across judges in vote mode) so the caller can attribute the cost separately from branch-agent usage, without extending pydantic-ai-shields' :class:CostTracking API with a new cost category.