Tools, agent loop, durable runs

Overview

The core of the harness. Add the Tool interface with the Result{ForLLM, ForUser, IsError} split, a registry that emits provider tool definitions, two safe builtin tools, and the think→act→observe loop with an iteration cap, dangling-tool-call repair and error-as-tool-result. Then make runs durable: a runs row created by POST /v1/sessions/{id}/runs returning 202, executed by an in-process worker pool that serializes per session, observable by polling GET /v1/runs/{id}. No streaming yet — that is phase 6.

Requirements

Functional

• Tool interface: name, description, JSON Schema parameters, Execute(ctx, args) Result.
• Registry: register, lookup, ProviderDefs(allowlist []string) []provider.ToolDef, per-agent filtered view from the agent’s tools allowlist (empty allowlist = all registered tools).
• Builtins: get_time(timezone?) and http_fetch(url, max_bytes?) with a config-driven host allowlist, 5s timeout and a hard response cap.
• Loop: build system prompt and history, call the provider, push events through a provider.Accumulator while forwarding them to the sink, persist the assistant message, execute requested tools, persist tool-result messages, repeat while StopReason == ToolUse; stop at HANGO_MAX_ITERATIONS (default 20) with a terminal message explaining the cap.
• Tool failures (panic, error, timeout, unknown tool, invalid args) become ToolResultBlock{IsError:true} and the loop continues.
• `pause_turn` is not resumed. The adapter returns `ErrUnsupportedStop`, the loop fails the run with that error, and the message names the stop reason. There is no resume counter.
• Dangling tool calls: before any provider call, if the last assistant message has tool_use blocks without matching tool_result, synthesize error tool results and persist them.
• runs table with queued|running|succeeded|failed|cancelled; POST /v1/sessions/{id}/runs inserts and enqueues, returns 202 with the run; GET /v1/runs/{id} returns state, error, usage totals and the final assistant message id.
• Only one run per session executes at a time; a second request while one is active is queued behind it.
• **Ownership and leases.** `runs` carries `owner_id uuid` (the process instance that claimed it) and `lease_expires_at timestamptz`. A worker claims work only through a conditional update — `UPDATE runs SET state='running', owner_id=$me, lease_expires_at=now()+$ttl, started_at=coalesce(started_at, now()) WHERE id=$1 AND state='queued' RETURNING *` — so two workers, or a worker and a restarted process, can never both execute one run: exactly one `UPDATE` returns a row. The worker renews the lease on a heartbeat every `HANGO_RUN_HEARTBEAT_INTERVAL` (default `lease_ttl/3`). There is no separate `MarkRunning`: the claim *is* the transition to `running`. `MarkTerminal`, `RequestCancel` and the approval transitions of phase 15 are all conditional on both the expected current state **and** `owner_id`, so a process that lost its lease cannot overwrite the state of a run someone else now owns.
• **Restart stance: requeue and resume at a turn boundary.** Reconciliation at startup (and on a periodic sweep) reclaims only runs whose `lease_expires_at < now()`, moving `running → queued`, clearing `owner_id`, and leaving `runs.resume_turn` pointing at the last completed turn (phase 7 adds `runs.state_json` for middleware state, which is written at the same boundary). A requeued run restarts from that turn boundary, not from the beginning: everything persisted up to the boundary (assistant message, tool results) is history it reads back, so no completed turn is re-executed. A turn interrupted mid-flight is re-driven, which means **tools must document their idempotency**: every builtin states in its Go doc whether re-execution is safe, and `http_fetch` (idempotent GET only) and `get_time` are trivially safe. Non-idempotent tools arriving later must be gated by phase 15 interrupts or made idempotent by a caller-supplied key.
• `HANGO_DRAIN_TIMEOUT` must be at least one full turn (documented default 120s) so a graceful shutdown lands on a turn boundary rather than mid-turn; a shorter value is legal but produces re-driven turns on restart, and the config validation warns about it.
• Enqueue is deduplicated: the scheduler's in-memory queue is keyed by run id and a re-enqueue of a run already queued or running is a no-op, so approval resume, the DB poller (phase 6) and reconciliation can all enqueue the same run without racing.
• Run authorization: a `member` may create, read and cancel runs only on sessions they created; `owner`/`admin` may act on any run in the tenant. The predicate comes from the shared `visibility(scope)` helper; denials on read are 404.

Non-functional

• Tool execution never panics into the loop: every Execute call is wrapped with recover.
• Each tool call has an individual timeout from config; the loop has an overall run deadline.
• The worker never holds a database connection across a provider call.
• Integration tests use provider.Fake plus real Postgres; no network.
• **Locked dependency rule: `internal/agent` imports nothing from `internal/run`.** The agent package defines the interfaces; `internal/run` implements them. Concretely: `agent.EventSink` is defined in `internal/agent` and implemented by `run.Emitter`; the subagent lane gate (phase 9) is `agent.Gate func(ctx context.Context) (release func(), error)` defined in `internal/agent` and supplied by `internal/app` from the scheduler. Any type that would need the reverse direction is a design error, not a reason to relax the rule.
• `Agent.Run`'s signature is finalized here and does not change again. Growth happens inside `RunOptions`, so phase 9's depth and parent-sink additions are field additions, not a signature break. Call sites to update when the signature was introduced: `internal/run/worker.go` (the only production caller), `internal/agent/loop_test.go`, `internal/run/scheduler_test.go`, and from phase 9 onward `internal/agent/mwsubagent.go`. There are no others.

Architecture

// internal/tool/tool.go
type Tool interface {
	Name() string
	Description() string
	Parameters() map[string]any // JSON Schema object
	Execute(ctx context.Context, args json.RawMessage) Result
}

type Result struct {
	ForLLM  string // what the model sees
	ForUser string // what a human/UI sees; falls back to ForLLM when empty
	IsError bool
}
// Red team: F15 — Result.Meta was never read by any phase; the offload wrapper (phase 8)
// records its path on the span and in its own state, so the field is deleted.

func Errorf(format string, a ...any) Result // IsError: true

// internal/tool/registry.go
type Registry struct{} // concurrency-safe, built once at startup

func (r *Registry) Register(t Tool) error            // rejects duplicate names
func (r *Registry) Get(name string) (Tool, bool)
func (r *Registry) View(allow []string) *View        // per-agent filtered view
func (v *View) ProviderDefs() []provider.ToolDef
func (v *View) Call(ctx context.Context, call provider.ToolUseBlock, timeout time.Duration) Result // recovers panics

// internal/agent/loop.go
type Agent struct {
	Def      store.Agent
	Provider provider.Provider
	Tools    *tool.View
	Store    store.Scoped
	Limits   Limits // MaxIterations, ToolTimeout, RunDeadline  // Red team: D4 — MaxResumes removed
}

type Status uint8 // StatusDone, StatusFailed, StatusCancelled // StatusAwaitingApproval added in phase 15

type Result struct {
	Status        Status
	StopReason    provider.StopReason
	Iterations    int
	Usage         provider.Usage
	FinalMessage  *store.Message
	Err           error
}

// Red team: F5 — final signature; later phases add fields to RunOptions, never parameters.
type RunOptions struct {
	Depth      int        // 0 for a top-level run; phase 9 sets it for children
	ParentSink EventSink  // nil at depth 0; phase 9 wraps the parent's sink with depth labels
	ResumeFrom int        // turn index to resume at; 0 means start from history. Red team: D3
}

func (a *Agent) Run(ctx context.Context, sessionID uuid.UUID, input []provider.Block, emit EventSink, opt RunOptions) (Result, error)

// EventSink is defined here and implemented in internal/run (never the reverse). // Red team: F5
type EventSink interface { // phase 6 replaces the no-op implementation with the broadcaster-backed emitter
	Emit(ctx context.Context, ev Event) error
}

// internal/run/scheduler.go
type Scheduler struct{} // in-process; single replica (documented limitation)

func (s *Scheduler) Enqueue(ctx context.Context, r store.Run) error // per-session FIFO, lane "main"; dedup by run id
func (s *Scheduler) Shutdown(ctx context.Context) error             // drain; phase 6 adds MarkDraining

// Red team: D3, F2 — the only way a worker starts executing a run.
// Returns (run, true) for exactly one caller; everyone else gets (zero, false).
func (s *RunStore) ClaimQueued(ctx context.Context, runID, ownerID uuid.UUID, lease time.Duration) (store.Run, bool, error)
func (s *RunStore) RenewLease(ctx context.Context, runID, ownerID uuid.UUID, lease time.Duration) error
func (s *RunStore) ReclaimExpired(ctx context.Context, limit int) (int, error) // running+expired lease -> queued

-- migrations/00003_runs.sql
create table runs (
  id uuid primary key,
  tenant_id uuid not null references tenants(id) on delete cascade,
  session_id uuid not null references sessions(id) on delete cascade,
  agent_id uuid not null references agents(id) on delete cascade,
  created_by uuid not null references users(id),
  lane text not null default 'main',
  state text not null check (state in ('queued','running','succeeded','failed','cancelled')),
  owner_id uuid,                    -- Red team: D3 — process instance holding the run
  lease_expires_at timestamptz,     -- Red team: D3 — only an expired lease may be reclaimed
  resume_turn int not null default 0, -- Red team: D3 — turn boundary a requeued run restarts at
  input jsonb not null,
  error text not null default '',
  stop_reason text not null default '',
  iterations int not null default 0,
  usage jsonb,
  final_message_id uuid references messages(id),
  trace_id text not null default '',
  span_id text not null default '',
  cancel_requested_at timestamptz,
  started_at timestamptz,
  finished_at timestamptz,
  created_at timestamptz not null default now()
);
create index runs_tenant_session_created_idx on runs (tenant_id, session_id, created_at desc);
create index runs_state_idx on runs (state) where state in ('queued','running');
-- Red team: D3 — reclaim sweep and the phase-6 queued-run poller both read this index
create index runs_lease_idx on runs (state, lease_expires_at) where state = 'running';
create index runs_queued_created_idx on runs (state, created_at) where state = 'queued';

Instrumentation

• Root span per run: invoke_agent {agent name}, kind INTERNAL, created by the worker as the root of a new trace (the HTTP request that created the run is a separate trace linked by a span link, because the run outlives the request). Attributes: gen_ai.operation.name=invoke_agent, gen_ai.agent.name, gen_ai.agent.id, gen_ai.conversation.id=<session_id>, hango.tenant.id, hango.agent.id, hango.session.id, hango.run.id, hango.lane.
• Langfuse trace-level attributes on the root span: langfuse.observation.type=agent, langfuse.trace.name=<agent name>, langfuse.session.id=<hango session_id>, langfuse.user.id=<run.created_by>, langfuse.trace.tags (agent slug, lane), langfuse.trace.metadata.tenant_id, ...metadata.agent_id, ...metadata.run_id. This is what makes any failure locatable by tenant and run in the Langfuse UI.
• Turn span: chain observation per iteration (hango.turn attribute, langfuse.observation.type=chain) parenting that turn’s chat span and its execute_tool spans.
• Tool span: execute_tool {tool} per call with gen_ai.operation.name=execute_tool, gen_ai.tool.name, gen_ai.tool.call.id, langfuse.observation.type=tool, error.type on failure, and input/output recorded only under HANGO_TRACE_LLM_CONTENT.
• Metrics (all names declared in internal/telemetry/metrics_names.go): gen_ai.invoke_agent.duration, gen_ai.invoke_agent.inference_calls, gen_ai.invoke_agent.tool_calls, gen_ai.execute_tool.duration (attribute gen_ai.tool.name), hango.runs.by_state gauge (attributes state, hango.tenant.id), hango.run.queue_depth gauge (attribute hango.lane), hango.tool.errors counter (attributes gen_ai.tool.name, error.type), hango.run.iterations histogram, hango.run.claims counter (attribute outcome = won|lost) and hango.run.reclaimed counter.
• Span events on the run root span: run.claimed (with owner id and lease deadline), run.lease_renewed at debug, run.reclaimed (with the previous owner and the resume turn).
• Run row stores trace_id (and the root span_id) so GET /v1/runs/{id} returns them and a user can paste the trace id straight into Tempo or open the Langfuse trace.
• Log fields: the worker installs a context logger bound to hango.tenant.id, hango.agent.id, hango.session.id, hango.run.id, so every line emitted anywhere inside the loop is filterable by run.

Related Code Files

Create

• internal/tool/tool.go, internal/tool/registry.go, internal/tool/schema.go (helpers to build JSON Schema objects), internal/tool/registry_test.go.
• internal/tool/builtin/gettime.go, internal/tool/builtin/httpfetch.go, plus tests (httptest server for fetch).
• internal/agent/loop.go, internal/agent/prompt.go (short dynamic system prompt assembly), internal/agent/repair.go (dangling tool-call repair), internal/agent/loop_test.go.
• internal/run/scheduler.go, internal/run/worker.go, internal/run/registry.go (in-memory active-run map), internal/run/scheduler_test.go.
• internal/agent/trace.go — run/turn/tool span helpers and the agent-level metric instruments.
• internal/agent/trace_test.go — in-memory span-recorder test asserting the invoke_agent → chain → chat/execute_tool hierarchy and the Langfuse attribute set.
• internal/store/pg/run.go, internal/store/pg/run_test.go.
• internal/server/handlers_runs.go.
• migrations/00003_runs.sql.
• internal/run/reconcile.go, internal/run/reconcile_test.go — the lease-expiry reclaim sweep.
• docs/learning/05-tools-loop-and-runs.md.

Modify

• internal/store/store.go, internal/store/types.go — Run, RunStore.
• internal/store/pg/tenancy_test.go — add the runs cross-tenant and intra-tenant cases.
• internal/config/config.go — HANGO_MAX_ITERATIONS, HANGO_TOOL_TIMEOUT, HANGO_RUN_DEADLINE, HANGO_RUN_WORKERS, HANGO_HTTP_FETCH_ALLOWED_HOSTS, HANGO_MAX_QUEUED_RUNS_PER_SESSION, HANGO_RUN_LEASE_TTL (default 90s), HANGO_RUN_HEARTBEAT_INTERVAL (default lease/3), HANGO_RUN_RECLAIM_INTERVAL.
• internal/app/app.go — build registry, register builtins, build scheduler, start workers, register scheduler shutdown.
• internal/server/server.go, internal/server/handlers_agents.go (validate tools against the registry), docs/system-architecture.md, docs/learning/README.md.

Review Checkpoints

Implementation Steps

1. Write internal/tool/tool.go and schema.go. schema.Object(schema.Field("url", "string", "…", true), …) keeps JSON Schema literals out of tool files.
2. Write registry.go. View.Call looks up the tool, unmarshals nothing (tools own their args), applies the per-call timeout via context.WithTimeout, wraps Execute in a function with defer recover() converting a panic into Result{IsError:true} with the panic message and a logged stack.
3. Write the two builtins. http_fetch resolves the URL, rejects non-http(s), rejects hosts not in the allowlist, rejects private/loopback IPs after resolution (SSRF), streams at most max_bytes (default 64KiB) and returns the body plus status.
4. Write internal/agent/prompt.go: the system prompt is the agent’s system_prompt plus, later, prompt-contributor fragments. Nothing authored by hango. Return []provider.Block so cache control can be attached in phase 10.
5. Write internal/agent/repair.go: given []provider.Message, find unmatched tool_use ids in the final assistant message and produce synthetic ToolResultBlock{IsError:true, Content:"tool call interrupted"}. Persist them through the message store before the first provider call.
6. Write internal/agent/loop.go:
   • Load History, run repair, append the caller’s input blocks as a user message. When RunOptions.ResumeFrom > 0, skip forward to that turn boundary instead of re-appending the input.
   • Iterate: build provider.Request (model, system, messages, Tools.ProviderDefs(), max tokens), call Stream, push every event into a provider.Accumulator while forwarding it to the sink, persist the assistant message with stop reason and usage. On a mid-stream error, persist Accumulator.Partial() and fail the turn — never silently drop what arrived.
   • If StopReason == ToolUse: for each ToolUseBlock, call Tools.Call, build one user message containing all ToolResultBlocks in request order, persist it, record the completed turn in runs.resume_turn, continue.
   • If StopReason == PauseTurn: the provider already returned ErrUnsupportedStop; fail the run with that error.
   • If StopReason == ContextExceeded: fail with a typed error for now; phase 10 turns this into compaction-and-retry.
   • Stop on EndTurn|MaxTokens|StopSequence|Refusal, on the iteration cap, or on ctx.Err().
   • Accumulate usage across turns.
7. Write RunStore: Create, Get, ListBySession, ClaimQueued, RenewLease, MarkTerminal(...), RequestCancel, ReclaimExpired. Every mutation is a single conditional UPDATE ... WHERE id=$1 AND state=$expected AND owner_id=$me RETURNING and returns ErrStateConflict when it affects no row; no method reads-then-writes.
8. Write internal/run/scheduler.go: a bounded worker pool of HANGO_RUN_WORKERS goroutines reading from a lane queue, plus a per-session FIFO map so only one run per session is in flight. Enqueue inserts nothing (the handler already created the row), dedups by run id, and pushes the run id. Keep an in-memory registry mapping run id → cancel func.
9. Write internal/run/worker.go: pop a run id, claim it with ClaimQueued (skip silently when the claim returns false — someone else owns it), rebuild the tenant scope from the claimed row (the request context is gone): load the creator’s current user row by run.TenantID/run.CreatedBy and build store.Scope{TenantID, UserID, Role} from its current role. When the creator is gone or its status is not active, MarkTerminal the run as failed with error creator_inactive and stop; the run never executes under rights the creator no longer has. Then load the agent, build the Agent, start the heartbeat via safego, execute with RunOptions{ResumeFrom: run.ResumeTurn}, MarkTerminal gated on owner_id. Finalization runs on a background context derived with context.WithoutCancel so a cancelled run still persists its partial state.
10. Write internal/server/handlers_runs.go: POST /v1/sessions/{id}/runs authorizes the session against the scope (creator-scoped for member), creates the run, enqueues, returns 202 + Location: /v1/runs/{id}. GET /v1/runs/{id} returns the run. Reject a new run when the session already has HANGO_MAX_QUEUED_RUNS_PER_SESSION pending.
11. Write the reconciliation sweep (safego ticker at HANGO_RUN_RECLAIM_INTERVAL, plus one pass at startup): ReclaimExpired moves running runs whose lease expired back to queued, clears owner_id, leaves resume_turn untouched, and re-enqueues them. Runs whose lease is still valid are left alone — that is what makes a rolling restart safe.
12. Tests: registry duplicate/panic/timeout; loop with provider.Fake scripted to (a) text only, (b) one tool call then text, (c) tool error then recovery, (d) iteration-cap exhaustion, (e) pause_turn producing ErrUnsupportedStop and a failed run; run lifecycle integration test; per-session serialization test; two workers racing to claim one queued run: exactly one executes (asserted with a counting fake provider); a MarkTerminal from a stale owner returns ErrStateConflict and does not change the row; a killed-mid-turn run is reclaimed after lease expiry and resumes at its turn boundary without repeating a completed turn; runs cross-tenant and intra-tenant cases.
13. Write internal/agent/trace.go and wire it: the worker starts the root invoke_agent span with a span link to the creating request’s span context (persisted on the run row at creation), stores trace_id/span_id on the run, opens a chain span per turn, and the tool path opens execute_tool spans. Register the agent-level metric instruments (names from internal/telemetry/metrics_names.go) and update hango.runs.by_state and hango.run.queue_depth from the scheduler.
14. Call `SetAuditAction(c, "run.create", "run", runID)` in the run-creation handler, and record the agent id, session id and lane in `detail`. The audit row is written after the handler returns, so it carries the same `trace_id` the run row stores, which is what lets a reviewer move from "who started this run" to the run's own trace in one hop.
15. Update docs and the learning doc.

Learning Doc Outline

docs/learning/05-tools-loop-and-runs.md must contain:

• Concept: the think→act→observe loop, why it terminates, and what the stop reason actually controls.
• Design choices: Result.ForLLM/ForUser split, error-as-tool-result, panic recovery at the registry boundary, per-agent tool views, iteration cap, dangling-tool-call repair, durable run row + 202 + poll, and the ownership lease.
• Ownership, leases and turn-boundary resume: why "claim by conditional UPDATE" is the whole concurrency design, why a lease beats a heartbeat table, what `resume_turn` guarantees, and why tool idempotency is a documented property rather than an assumption.
• The one-directional package rule (`internal/agent` never imports `internal/run`) and how `EventSink` plus an injected gate function keep it true.
• Trade-offs: in-request execution vs durable worker (disconnect, graceful shutdown, resumability, HITL); per-session serialization vs optimistic concurrency; tool timeout placement.
• The prompt discipline decision: no authored base prompt, policy in tool descriptions.
• goclaw comparison: Loop.Run with a V2 monolith and a V3 stage pipeline behind a feature flag, iteration cap 20, FinalizeStage on a background context, toolloop.go repeat detection, Tool interface with Result.ForLLM/ForUser, 5-layer tool policy engine, lane scheduler with per-session queues.
• How to debug this layer: from a failed run id, GET /v1/runs/{id} returns trace_id; paste it into Tempo to see the invoke_agent root, which turn failed, which chat or execute_tool span carries the error, and the exact tool arguments; the same run opens in Langfuse filtered by metadata.run_id with the generation and tool observations in order; Loki filtered by hango.run.id gives the full log narrative.
• deepagents comparison: the loop is LangGraph’s create_agent; tools are the middleware-provided surface; PatchToolCallsMiddleware performs dangling-tool-call repair at before_agent; recursion_limit is effectively unbounded.
• Why hango diverges: one loop implementation (never two behind a flag), repair as an explicit step, a durable run row instead of an in-request goroutine, a simple allowlist instead of a policy engine.
• Exercises: add a third tool; make a tool panic and watch the loop survive; set HANGO_MAX_ITERATIONS=1 and read the terminal message; kill -9 the process mid-run and watch the lease expire, the sweep requeue and the run resume.
• Time budget: 17h implementation + 3h learning doc.

Success Criteria

• go build ./... && go vet ./... && go test ./... pass.
• Loop tests cover text-only, tool call, tool error, iteration cap, and pause_turn failing the run with ErrUnsupportedStop; each asserts the persisted message sequence.
• Two workers claiming the same queued run: exactly one ClaimQueued returns a row, exactly one provider call is made, and hango.run.claims{outcome="lost"} increments once.
• A worker killed mid-turn leaves the run running with an expiring lease; after expiry the sweep requeues it, a second worker claims it, and the run completes without re-executing the turns already persisted (counting fake tool asserts each completed tool ran once).
• MarkTerminal from a process whose lease was reclaimed returns ErrStateConflict and leaves the row unchanged.
• A member cannot create, read or cancel a run on another user’s session (404); owner/admin can.
• An owner’s run on a member’s session executes and loads that session; a queued run whose creator was disabled before the claim ends failed with error creator_inactive and makes no provider call (tests).
• Creating a run writes an audit_events row with action run.create, target_kind = 'run' and the run’s trace_id (integration test).
• A panicking tool produces IsError tool result and the run still reaches succeeded.
• POST /v1/sessions/{id}/runs returns 202 with Location; polling GET /v1/runs/{id} transitions queued → running → succeeded.
• Per-session serialization test: two runs enqueued on one session execute strictly sequentially (asserted by overlapping timestamps being impossible).
• go test ./internal/store/pg/ -run TestCrossTenant includes runs.
• http_fetch rejects a non-allowlisted host and a loopback address; tests assert both.
• Restarting the process requeues a previously running run whose lease has expired (never fails it) and leaves a run whose lease is still held by another live process untouched.
• GET /v1/runs/{id} returns a trace_id that resolves in Tempo to an invoke_agent root span with chain, chat and execute_tool children in the right order.
• The same run appears in Langfuse as a trace named after the agent, with session.id, user.id and metadata.tenant_id/agent_id/run_id set.
• hango.runs.by_state, hango.run.queue_depth and hango.tool.errors series exist in Prometheus after a test run.

Risk Assessment

Rollback

goose down one step drops runs. The loop and tool packages are additive; unmounting the run routes and not starting the scheduler in internal/app returns the server to phase 4 behaviour with messages still appendable via the API.