ToolOutcomes — Typed Outcome Union

File: brain/agent/tool_outcomes.py

---

Overview

Every tool call produces exactly one ToolOutcome. This is a closed union of five frozen dataclasses that exhaustively cover every possible result of dispatching a single tool.

Using typed outcomes instead of dict[str, Any] makes control flow explicit, enables exhaustive pattern matching, and catches missing cases at development time.

---

The Five Outcome Types

ToolExecutionResult — Successful execution

@dataclass(frozen=True)
class ToolExecutionResult:
    call_id: str
    tool_name: str
    output: dict[str, Any]   # The tool's return value
    elapsed_ms: int
    was_coerced: bool = False  # True if ToolArgValidator coerced the arguments

Produced when the tool ran without error and the output fits inline (≤ TOOL_MESSAGE_MAX_CHARS = 12_000 characters).

ToolTimeout — Exceeded deadline

@dataclass(frozen=True)
class ToolTimeout:
    call_id: str
    tool_name: str
    deadline_s: float    # The absolute deadline that was violated
    elapsed_ms: int
    retryable: bool = True

retryable is read from tool_spec.metadata["retry_on_timeout"]. If False, the tool is added to state.blocked_tool_names for the rest of the turn.

ToolFailure — Exception or error response

@dataclass(frozen=True)
class ToolFailure:
    call_id: str
    tool_name: str
    error: str
    retryable: bool = True
    elapsed_ms: int = 0

Produced when the tool raises an exception or returns {"error": ...}. retryable=False triggers blocking.

ToolDenied — Pre-execution skip

@dataclass(frozen=True)
class ToolDenied:
    call_id: str
    tool_name: str
    reason: str     # See reasons table below
    details: str = ""

Produced before the tool even executes, by one of the 4 pre-execution gates.

reason	Produced by
"duplicate"	ReplayControl — idempotent tool with a prior successful call
"blocked"	Per-turn blocked list — tool produced a non-retryable failure earlier
"pre_hook"	on_pre_tool_use callback returned allow=False
"validation"	ToolArgValidator failed and repair attempts exhausted
"deadline"	TurnBudget.is_expired() was True before execution started
"write_denied"	Write confirmation callback returned False

ToolArtifactReference — Oversized output

@dataclass(frozen=True)
class ToolArtifactReference:
    call_id: str
    tool_name: str
    artifact_id: str    # Key in ArtifactStore
    summary: str        # ≤200 char preview
    size_bytes: int = 0

Produced when the serialized output exceeds 12,000 characters. The full content is stored in ArtifactStore; the model receives a compact reference message instructing it to use read_file to retrieve the full content.

---

Type Alias

ToolOutcome = Union[
    ToolExecutionResult,
    ToolTimeout,
    ToolFailure,
    ToolDenied,
    ToolArtifactReference,
]

---

Serialisation: outcome_to_model_content(outcome) → str

Converts any ToolOutcome to the JSON string placed in the tool-role message that the LLM receives as the result of its tool call.

Outcome type	Model sees
ToolExecutionResult	json.dumps(outcome.output)
ToolTimeout	{"status": "error", "error": "...", "timed_out": true, "retryable": ...}
ToolFailure	{"status": "error", "error": "...", "retryable": ...}
ToolDenied("duplicate")	{"warning": "duplicate_tool_call", "skipped": true}
ToolDenied("blocked")	{"warning": "non_retryable_tool_failure", "skipped": true}
ToolDenied("validation")	{"error": "argument_validation_failed", "details": "...", "hint": "..."}
ToolDenied("deadline")	{"error": "Turn deadline expired; cannot execute tool.", "timed_out": true}
ToolDenied(other)	{"error": "Blocked: ...", "blocked": true}
ToolArtifactReference	{"artifact_reference": "...", "summary": "...", "hint": "..."}

---

Helper Predicates

outcome_is_error(outcome)     → bool  # True for ToolTimeout or ToolFailure
outcome_is_retryable(outcome) → bool  # True if the outcome has retryable=True
outcome_blocks_tool(outcome)  → bool  # True if the tool should be blocked for this turn
                                       # (non-retryable ToolFailure or non-retryable ToolTimeout)

outcome_blocks_tool() is used by harness._apply_outcome_to_state() to populate state.blocked_tool_names.

---

Usage Pattern

outcome = bounded_executor.execute(call, state, web_mode=web_mode, ...)

# Inject tool result into message stream
messages.append({
    "role": "tool",
    "tool_call_id": call.id,
    "content": outcome_to_model_content(outcome),
})

# Update state
if outcome_blocks_tool(outcome):
    state.blocked_tool_names.add(call.name)

# Update replay control
if outcome_is_error(outcome):
    replay_control.record_failure(call.name, call.arguments, spec)
else:
    replay_control.record_success(call.name, call.arguments, spec)