OpenAgents

Commit

Harden asynchronous runtime recovery

6a812a22e994 · Christopher David · · parent 12939583ca39 

Harden asynchronous runtime recovery

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

Not deployed through the forge lane

No push, promotion, build, or deploy receipt references this commit (receipts are scanned over a bounded recent window). Changes shipped by full node replacement carry their proof in the release gate receipt instead.

Changed files

  • modified INVARIANTS.md
  • modified config/config.exs
  • added docs/operations/asynchronous-recovery-contract.md
  • modified lib/openagents/computer_agent_jobs.ex
  • modified lib/openagents/conversations.ex
  • modified lib/openagents/conversations/message.ex
  • modified lib/openagents/memory/semantic_index.ex
  • modified lib/openagents/memory/semantic_worker.ex
  • modified lib/openagents/providers/open_ai.ex
  • modified lib/openagents/runtime_config.ex
  • modified lib/openagents/turn_recovery.ex
  • modified lib/openagents/voice/open_ai/call_client.ex
  • modified lib/openagents/voice_recovery.ex
  • modified lib/openagents/work.ex
  • modified lib/openagents/work/delegation_server.ex
  • modified lib/openagents/work/job.ex
  • modified lib/openagents/work_recovery.ex
  • added priv/repo/migrations/20260820085203_harden_async_runtime_boundaries.exs
  • modified test/openagents/conversations_test.exs
  • added test/openagents/memory/semantic_worker_test.exs
  • modified test/openagents/providers/provider_contract_test.exs
  • modified test/openagents/semantic_recall_test.exs
  • modified test/openagents/tools/incident_lookup_test.exs
  • modified test/openagents/turn_provenance_test.exs
  • modified test/openagents/voice/open_ai/call_client_test.exs
  • modified test/openagents/voice_test.exs
  • modified test/openagents/work_job_test.exs
  • modified test/openagents_web/controllers/computer_agent_jobs_controller_test.exs
  • added test/support/memory/semantic_worker_test_provider.ex

Diff

29 files changed, +1226 -188

INVARIANTS.md modified +22 -6

@@ -424,6 +424,10 @@ asynchronous derivatives bound to the source message, exact conversation,

424 424 
content digest, model/version manifest, and active generation. Hybrid queries
425 425 
repeat the conversation and frozen snapshot predicates in PostgreSQL and admit
426 426 
only ready rows whose digest still matches the authoritative complete message.
427 
Each asynchronous embedding claim has a bounded lease and attempt number. A
428 
provider task must finish before the lease, and a replacement worker may
429 
reclaim only an expired running claim. Attempt fencing prevents a stale worker
430 
from publishing a late derivative.
427 431 
Correction, deletion, and rebuild invalidate derivative rows and produce
428 432 
append-only receipts; a new manifest generation cannot read an older one.
429 433 
Receipts cannot be edited or deleted while their authoritative conversation

@@ -575,7 +579,9 @@ Status: Current

575 579
576 580 
The user message, empty streaming assistant message, and turn record are
577 581 
inserted in one transaction. Completion, failure, and cancellation update both
578 
assistant-message and turn terminal state.
582 
assistant-message and turn terminal state. PostgreSQL applies a one-megabyte
583 
hard ceiling to every message, and the configured, lower assistant-message
584 
ceiling applies to the complete accumulated stream rather than each delta.
579 585
580 586 
Evidence: `OpenAgents.Conversations.create_turn/2`, `finish_turn/5`, and turn tests.
581 587

@@ -596,10 +602,13 @@ Status: Current

596 602
597 603 
Active records left by a runtime restart are marked failed during application
598 604 
startup. A response is never left permanently presented as in progress without
599 
an executing turn process.
605 
an executing turn process. Recovery records the bounded provider-neutral
606 
`runtime_restarted` code, fails every active provider and tool step in the same
607 
transaction, and is idempotent.
600 608
601 609 
Evidence: `OpenAgents.Conversations.recover_interrupted_turns/0`, the
602 
`OpenAgents.TurnRecovery` application child, and startup-recovery test.
610 
`OpenAgents.TurnRecovery` application child, and the process-death recovery
611 
test in `OpenAgents.TurnProvenanceTest`.
603 612
604 613 
### TURN-005 — Tool continuations are serial, bounded, and commit-first
605 614

@@ -642,7 +651,9 @@ event shapes. Adapters emit typed OpenAgents-domain lifecycle, text, tool-call,

642 651 
usage, completion, failure, and cancellation events. A response ID is persisted
643 652 
when announced, and matching explicit completion is required; stream closure
644 653 
alone cannot produce a completed turn. Provider-specific events, credentials,
645 
and raw errors never reach the receipt or browser.
654 
and raw errors never reach the receipt or browser. Response creation is a
655 
non-idempotent mutation and the adapter sends it exactly once; continuation is
656 
an explicit host decision backed by a committed tool outcome.
646 657
647 658 
Evidence: `OpenAgents.Providers.ProviderEvent`, `OpenAgents.Providers.OpenAI`,
648 659 
`OpenAgents.Providers.OpenAI.StreamDecoderTest`, `OpenAgents.Providers.Test`, and

@@ -977,7 +988,12 @@ ends the job as explicit `budget_exhausted`. Every terminal path (`completed`,

977 988 
streamed report text is persisted as it arrives, and a job that dies before a
978 989 
narrative gets an honest host summary of its committed step evidence.
979 990 
PostgreSQL constrains status transitions and makes terminal jobs and terminal
980 
steps immutable. Startup recovery RESUMES orphaned active jobs (#97): it
991 
steps immutable. A delegation additionally binds one account-owned machine,
992 
its admission-time authority snapshot, its bounded execution budget, and its
993 
immutable request. Only the generation-fenced ACP session ID may change after
994 
admission; workers read machine, agent, working directory, and wall-clock
995 
authority from the immutable fields. Startup recovery RESUMES orphaned active
996 
jobs (#97): it
981 997 
restarts each job's supervised worker, which re-claims through the generation
982 998 
fence and continues — a delegation by its durably checkpointed ACP session id,
983 999 
deep work from its committed evidence. A job whose cluster singleton is still

@@ -992,7 +1008,7 @@ best-effort projection that can never rewrite the committed terminal state.

992 1008
993 1009 
Evidence: `OpenAgents.Work`, `OpenAgents.Work.Job`, `OpenAgents.Work.JobStep`,
994 1010 
`OpenAgents.Work.JobServer`, `OpenAgents.WorkRecovery`, `OpenAgents.Tools.DeepWork`, the
995 
`create_work_jobs` migration triggers, `OpenAgents.WorkJobTest`, and
1011 
work-job migration triggers, `OpenAgents.WorkJobTest`, and
996 1012 
`OpenAgents.DeepWorkToolLoopTest`.
997 1013
998 1014 
### SELF-EDIT-001 — Every behavior change is anchored to a pushed commit (2026-08-19)
config/config.exs modified +3 -1

@@ -82,7 +82,9 @@ config :openagents,

82 82 
    model_version: "2024-01",
83 83 
    dimensions: 64,
84 84 
    batch_size: 10,
85 
    poll_interval_ms: 2_000
85 
    poll_interval_ms: 2_000,
86 
    provider_timeout_ms: 15_000,
87 
    lease_ms: 30_000
86 88 
  ],
87 89 
  experience_memory: [
88 90 
    enabled: false,
docs/operations/asynchronous-recovery-contract.md added +96

@@ -0,0 +1,96 @@

1 
# Asynchronous recovery contract
2
3 
Date: 2026-08-20
4
5 
Status: Gate 8 implementation contract
6
7 
This contract defines what the application must do when a runtime process dies
8 
after durable work has begun. PostgreSQL is authority. Processes, registry
9 
entries, PubSub messages, browser state, and provider streams are projections
10 
that may disappear or repeat.
11
12 
## Shared rules
13
14 
- Persist identity, ownership, scope, budget, and an execution fence before an
15 
  external effect begins.
16 
- Recovery may resume only when the external operation has a durable
17 
  idempotency or reattachment contract. Otherwise it must terminate the
18 
  durable row honestly and require a new user action.
19 
- Duplicate delivery must return or extend the existing durable record; it
20 
  must not repeat a completed mutation.
21 
- A stale process may not commit after a newer owner has acquired the durable
22 
  fence.
23 
- Every input, stream frame, accumulated projection, tool request, tool result,
24 
  report, and exported representation has a byte, item, time, or count bound.
25 
- Provider-specific errors and private content do not cross the adapter or log
26 
  boundary. Durable failures use bounded application error codes.
27 
- Recovery tests use supervised processes, process monitors, synchronous state
28 
  barriers, and durable-state assertions. Fixed sleeps are not correctness
29 
  mechanisms.
30
31 
## Text turns
32
33 
A text turn has no provider reattachment identity that can safely continue an
34 
arbitrary interrupted stream. If its `TurnServer` dies, `TurnRecovery` changes
35 
the active turn, assistant message, receipt, provider step, and tool steps to
36 
an explicit terminal interrupted/failed state in one transaction. It does not
37 
invent completion, usage, or provider output. Re-running recovery is a no-op.
38
39 
Provider response creation is not retried by the HTTP adapter. The host may
40 
start a new response only through the explicit, bounded tool-continuation
41 
protocol whose durable tool outcome is already committed.
42
43 
## Voice sessions
44
45 
A live Realtime transport cannot be reconstructed from PostgreSQL. If its
46 
runtime disappears, `VoiceRecovery` or the restarted `SessionServer` fails the
47 
admitted generation and fences all late events. Final transcript, response,
48 
usage, tool-step, and recording evidence already committed remains durable;
49 
ephemeral transcript deltas do not become final evidence. Typed chat remains
50 
available.
51
52 
Voice-call admission is not retried by the HTTP adapter. Sideband reconnect is
53 
bounded within the same admitted generation and does not create another call.
54
55 
## Work jobs
56
57 
Each delegated machine job durably binds the owner, conversation, machine,
58 
admission-time authority snapshot, and budget snapshot. Each claim increments
59 
the PostgreSQL generation fence. The worker reads the machine, agent, working
60 
directory, and wall-clock limit from the immutable admission fields. A database
61 
trigger makes the delegation request immutable except for its fenced ACP
62 
session checkpoint. `WorkRecovery` restarts an active singleton; the new worker
63 
adopts the row and, when a durable ACP session ID exists, reattaches that
64 
external session. A stale generation cannot checkpoint or finish the job. If a
65 
worker cannot be started, recovery writes an honest interrupted report and
66 
terminal state.
67
68 
Deep-work and coding jobs use the same generation fence and persist tool steps
69 
before execution. A completed provider call ID or tool invocation is never
70 
executed again merely because a delivery repeats.
71
72 
## Semantic derivatives
73
74 
Semantic embeddings are rebuildable derivatives, so retrying the computation
75 
is safe. Claiming a job sets a bounded lease. A second worker may reclaim only
76 
a running job whose lease expired. Provider execution has a shorter hard
77 
timeout than the lease, and embedding persistence rechecks the active manifest
78 
and authoritative message digest. Provider failure becomes a bounded terminal
79 
failure; a process or database failure leaves the leased job reclaimable. A
80 
source mutation or deletion invalidates the job and derivative before a stale
81 
result can become authoritative.
82
83 
## Required direct evidence
84
85 
- Kill a live `TurnServer`, run `TurnRecovery`, and assert every durable active
86 
  row becomes terminal without fabricated completion.
87 
- Start `VoiceRecovery` over an admitted active generation and assert the
88 
  runtime is ignored only after the durable session is failed.
89 
- Kill or orphan a work worker at durable checkpoints, run `WorkRecovery`, and
90 
  assert generation-fenced adoption or honest terminal fallback.
91 
- Exercise `SemanticWorker` through provider failure, contained drain/database
92 
  failure, and worker death followed by expired-lease reclamation.
93
94 
The broader Gate 8 suites continue to own cancellation, malformed provider
95 
events, reconnect, duplicate delivery, memory isolation, data rights, voice
96 
recording, machine credentials, and harmless delegated-work behavior.
lib/openagents/computer_agent_jobs.ex modified +18 -1

@@ -50,12 +50,29 @@ defmodule OpenAgents.ComputerAgentJobs do

50 50 
        }
51 51 
        |> put_optional("resume_session_id", resume_session_id)
52 52
53 
      authority_snapshot = %{
54 
        "machine_tier" => machine.tier,
55 
        "roots" => machine.roots,
56 
        "cwd" => cwd,
57 
        "agent_id" => agent_id,
58 
        "machine_name" => machine.name
59 
      }
60
61 
      budget_snapshot = %{
62 
        "wall_clock_ms" => @delegation_timeout_ms,
63 
        "maximum_prompt_bytes" => 8_000,
64 
        "maximum_report_bytes" => 8_000
65 
      }
66
53 67 
      attributes = %{
54 68 
        conversation_id: conversation.id,
55 69 
        owner_visitor_id: owner.id,
70 
        machine_id: machine.id,
56 71 
        surface: surface,
57 72 
        goal: String.slice("Delegate to #{agent_id} on #{machine.name}: #{prompt}", 0, 2_000),
58 
        delegation: delegation
73 
        delegation: delegation,
74 
        authority_snapshot: authority_snapshot,
75 
        budget_snapshot: budget_snapshot
59 76 
      }
60 77
61 78 
      Work.start_delegation(attributes)
lib/openagents/conversations.ex modified +9 -1

@@ -847,12 +847,19 @@ defmodule OpenAgents.Conversations do

847 847 
  end
848 848
849 849 
  def append_assistant_delta(%Turn{} = turn, delta) when is_binary(delta) do
850 
    maximum_bytes = Application.fetch_env!(:openagents, :maximum_message_bytes)
851
850 852 
    result =
851 853 
      Repo.transaction(fn ->
852 854 
        message = Repo.get_for_update!(Message, turn.assistant_message_id)
855 
        content = message.content <> delta
856
857 
        if byte_size(content) > maximum_bytes do
858 
          Repo.rollback(:assistant_message_limit_reached)
859 
        end
853 860
854 861 
        message
855 
        |> Message.changeset(%{content: message.content <> delta})
862 
        |> Message.changeset(%{content: content})
856 863 
        |> Repo.update!()
857 864 
      end)
858 865

@@ -951,6 +958,7 @@ defmodule OpenAgents.Conversations do

951 958 
            set: [
952 959 
              status: "failed",
953 960 
              error_message: "Sarah restarted before this response finished.",
961 
              error_code: "runtime_restarted",
954 962 
              completed_at: now,
955 963 
              updated_at: now
956 964 
            ]
lib/openagents/conversations/message.ex modified +15

@@ -6,6 +6,7 @@ defmodule OpenAgents.Conversations.Message do

6 6
7 7 
  @roles ~w(user assistant system)
8 8 
  @statuses ~w(streaming complete failed cancelled)
9 
  @maximum_content_bytes 1_048_576
9 10
10 11 
  @primary_key {:id, :binary_id, autogenerate: true}
11 12 
  @foreign_key_type :binary_id

@@ -46,6 +47,7 @@ defmodule OpenAgents.Conversations.Message do

46 47 
    |> validate_inclusion(:role, @roles)
47 48 
    |> validate_inclusion(:status, @statuses)
48 49 
    |> validate_inclusion(:modality, ~w(text voice))
50 
    |> validate_content_bound()
49 51 
    |> validate_voice_provenance()
50 52 
    |> foreign_key_constraint(:conversation_id)
51 53 
    |> foreign_key_constraint(:voice_session_id)

@@ -54,6 +56,19 @@ defmodule OpenAgents.Conversations.Message do

54 56 
    )
55 57 
  end
56 58
59 
  defp validate_content_bound(changeset) do
60 
    case get_field(changeset, :content) do
61 
      content when is_binary(content) and byte_size(content) <= @maximum_content_bytes ->
62 
        changeset
63
64 
      content when is_binary(content) ->
65 
        add_error(changeset, :content, "exceeds #{@maximum_content_bytes} bytes")
66
67 
      _invalid ->
68 
        changeset
69 
    end
70 
  end
71
57 72 
  defp validate_voice_provenance(changeset) do
58 73 
    modality = get_field(changeset, :modality)
59 74
lib/openagents/memory/semantic_index.ex modified +110 -61

@@ -88,18 +88,20 @@ defmodule OpenAgents.Memory.SemanticIndex do

88 88 
    )
89 89 
  end
90 90
91 
  @spec process_next(module()) :: {:ok, :empty | :completed | :failed | :invalidated}
92 
  def process_next(provider) when is_atom(provider) do
93 
    case claim_job() do
91 
  @spec process_next(module(), keyword()) :: {:ok, :empty | :completed | :failed | :invalidated}
92 
  def process_next(provider, options \\ []) when is_atom(provider) and is_list(options) do
93 
    lease_ms = Keyword.get(options, :lease_ms, 30_000)
94
95 
    case claim_job(lease_ms) do
94 96 
      nil -> {:ok, :empty}
95 
      job -> execute_job(job, provider)
97 
      job -> execute_job(job, provider, options)
96 98 
    end
97 99 
  end
98 100
99 
  @spec process_all(module(), pos_integer()) :: map()
100 
  def process_all(provider, limit \\ 100) do
101 
  @spec process_all(module(), pos_integer(), keyword()) :: map()
102 
  def process_all(provider, limit \\ 100, options \\ []) do
101 103 
    Enum.reduce_while(1..limit, %{completed: 0, failed: 0, invalidated: 0}, fn _index, counts ->
102 
      case process_next(provider) do
104 
      case process_next(provider, options) do
103 105 
        {:ok, :empty} -> {:halt, counts}
104 106 
        {:ok, status} -> {:cont, Map.update!(counts, status, &(&1 + 1))}
105 107 
      end

@@ -134,12 +136,16 @@ defmodule OpenAgents.Memory.SemanticIndex do

134 136 
  def vector_literal(values) when is_list(values),
135 137 
    do: "[" <> Enum.map_join(values, ",", &float_literal/1) <> "]"
136 138
137 
  defp claim_job do
139 
  defp claim_job(lease_ms) do
140 
    now = DateTime.utc_now()
141
138 142 
    Repo.transaction(fn ->
139 143 
      job =
140 144 
        Repo.one(
141 145 
          from(job in SemanticJob,
142 
            where: job.status == "pending" and job.available_at <= ^DateTime.utc_now(),
146 
            where:
147 
              job.status in ["pending", "running"] and
148 
                job.available_at <= ^now,
143 149 
            order_by: [asc: job.inserted_at, asc: job.id],
144 150 
            lock: "FOR UPDATE SKIP LOCKED",
145 151 
            limit: 1

@@ -151,7 +157,10 @@ defmodule OpenAgents.Memory.SemanticIndex do

151 157 
        |> SemanticJob.lifecycle_changeset(%{
152 158 
          status: "running",
153 159 
          attempts: job.attempts + 1,
154 
          started_at: DateTime.utc_now()
160 
          error_code: nil,
161 
          started_at: now,
162 
          completed_at: nil,
163 
          available_at: DateTime.add(now, lease_ms, :millisecond)
155 164 
        })
156 165 
        |> update_or_rollback()
157 166 
      end

@@ -162,24 +171,24 @@ defmodule OpenAgents.Memory.SemanticIndex do

162 171 
    end
163 172 
  end
164 173
165 
  defp execute_job(job, provider) do
174 
  defp execute_job(job, provider, options) do
166 175 
    message = Repo.get(Message, job.message_id)
167 176
168 177 
    cond do
169 178 
      is_nil(message) or message.status != "complete" -> finish_invalidated(job)
170 179 
      Canonical.sha256(message.content) != job.content_digest -> finish_invalidated(job)
171 
      true -> call_provider(job, message, provider)
180 
      true -> call_provider(job, message, provider, options)
172 181 
    end
173 182 
  end
174 183
175 
  defp call_provider(job, message, provider) do
184 
  defp call_provider(job, message, provider, options) do
176 185 
    config = %{
177 186 
      model_id: job.model_id,
178 187 
      model_version: job.model_version,
179 188 
      dimensions: job.dimensions
180 189 
    }
181 190
182 
    case provider.embed(message.content, config) do
191 
    case invoke_provider(provider, message.content, config, options) do
183 192 
      {:ok, embedding} when is_list(embedding) and length(embedding) == job.dimensions ->
184 193 
        persist_embedding(job, message, embedding)
185 194

@@ -194,48 +203,69 @@ defmodule OpenAgents.Memory.SemanticIndex do

194 203 
    end
195 204 
  end
196 205
206 
  defp invoke_provider(provider, content, config, options) do
207 
    timeout_ms = Keyword.get(options, :provider_timeout_ms, 15_000)
208
209 
    task =
210 
      Task.Supervisor.async_nolink(OpenAgents.ProviderTaskSupervisor, fn ->
211 
        provider.embed(content, config)
212 
      end)
213
214 
    case Task.yield(task, timeout_ms) || Task.shutdown(task, :brutal_kill) do
215 
      {:ok, result} -> result
216 
      {:exit, _reason} -> {:error, :embedding_provider_failed}
217 
      nil -> {:error, :embedding_provider_timeout}
218 
    end
219 
  end
220
197 221 
  defp persist_embedding(job, message, embedding) do
198 222 
    Repo.transaction(fn ->
199 223 
      locked = Repo.get_for_update!(SemanticJob, job.id)
200 224 
      current_message = Repo.get!(Message, message.id)
201 225 
      active = active_manifest()
202 226
203 
      if locked.status != "running" or is_nil(active) or active.id != locked.manifest_id or
204 
           Canonical.sha256(current_message.content) != locked.content_digest do
205 
        finish_invalidated_locked(locked)
206 
      else
207 
        id = Ecto.UUID.generate()
208 
        vector = vector_literal(embedding)
209
210 
        _result =
211 
          Repo.query!(
212 
            "INSERT INTO message_semantic_embeddings (id,message_id,conversation_id,manifest_id,generation,model_id,model_version,dimensions,content_digest,status,embedding,inserted_at,updated_at) VALUES ($1::text::uuid,$2::text::uuid,$3::text::uuid,$4::text::uuid,$5,$6,$7,$8,$9,'ready',$10::text::vector,now(),now()) ON CONFLICT (message_id,generation) DO UPDATE SET content_digest=EXCLUDED.content_digest, status='ready', embedding=EXCLUDED.embedding, updated_at=now()",
213 
            [
214 
              id,
215 
              locked.message_id,
216 
              locked.conversation_id,
217 
              locked.manifest_id,
218 
              locked.generation,
219 
              locked.model_id,
220 
              locked.model_version,
221 
              locked.dimensions,
222 
              locked.content_digest,
223 
              vector
224 
            ]
225 
          )
226
227 
        locked
228 
        |> SemanticJob.lifecycle_changeset(%{
229 
          status: "completed",
230 
          error_code: nil,
231 
          completed_at: DateTime.utc_now()
232 
        })
233 
        |> update_or_rollback()
227 
      cond do
228 
        locked.status != "running" or locked.attempts != job.attempts ->
229 
          :superseded
230
231 
        is_nil(active) or active.id != locked.manifest_id or
232 
            Canonical.sha256(current_message.content) != locked.content_digest ->
233 
          finish_invalidated_locked(locked)
234
235 
        true ->
236 
          id = Ecto.UUID.generate()
237 
          vector = vector_literal(embedding)
238
239 
          _result =
240 
            Repo.query!(
241 
              "INSERT INTO message_semantic_embeddings (id,message_id,conversation_id,manifest_id,generation,model_id,model_version,dimensions,content_digest,status,embedding,inserted_at,updated_at) VALUES ($1::text::uuid,$2::text::uuid,$3::text::uuid,$4::text::uuid,$5,$6,$7,$8,$9,'ready',$10::text::vector,now(),now()) ON CONFLICT (message_id,generation) DO UPDATE SET content_digest=EXCLUDED.content_digest, status='ready', embedding=EXCLUDED.embedding, updated_at=now()",
242 
              [
243 
                id,
244 
                locked.message_id,
245 
                locked.conversation_id,
246 
                locked.manifest_id,
247 
                locked.generation,
248 
                locked.model_id,
249 
                locked.model_version,
250 
                locked.dimensions,
251 
                locked.content_digest,
252 
                vector
253 
              ]
254 
            )
255
256 
          locked
257 
          |> SemanticJob.lifecycle_changeset(%{
258 
            status: "completed",
259 
            error_code: nil,
260 
            completed_at: DateTime.utc_now()
261 
          })
262 
          |> update_or_rollback()
234 263
235 
        :completed
264 
          :completed
236 265 
      end
237 266 
    end)
238 267 
    |> case do
268 
      {:ok, :superseded} -> {:ok, :invalidated}
239 269 
      {:ok, status} -> {:ok, status}
240 270 
      {:error, _reason} -> finish_failed(job, "embedding_persist_failed")
241 271 
    end

@@ -244,25 +274,44 @@ defmodule OpenAgents.Memory.SemanticIndex do

244 274 
  defp finish_failed(job, reason) do
245 275 
    error_code = reason |> String.replace(~r/[^a-z0-9_]/, "_") |> String.slice(0, 64)
246 276
247 
    job
248 
    |> SemanticJob.lifecycle_changeset(%{
249 
      status: "failed",
250 
      error_code: error_code,
251 
      completed_at: DateTime.utc_now()
252 
    })
253 
    |> Repo.update()
277 
    result =
278 
      Repo.transaction(fn ->
279 
        locked = Repo.get_for_update!(SemanticJob, job.id)
280
281 
        if locked.status == "running" and locked.attempts == job.attempts do
282 
          locked
283 
          |> SemanticJob.lifecycle_changeset(%{
284 
            status: "failed",
285 
            error_code: error_code,
286 
            completed_at: DateTime.utc_now()
287 
          })
288 
          |> update_or_rollback()
289 
        else
290 
          locked
291 
        end
292 
      end)
254 293
255 
    {:ok, :failed}
294 
    case result do
295 
      {:ok, %SemanticJob{status: "failed"}} -> {:ok, :failed}
296 
      {:ok, _superseded} -> {:ok, :invalidated}
297 
      {:error, reason} -> raise "semantic failure persistence failed: #{inspect(reason)}"
298 
    end
256 299 
  end
257 300
258 301 
  defp finish_invalidated(job) do
259 
    job
260 
    |> SemanticJob.lifecycle_changeset(%{
261 
      status: "invalidated",
262 
      error_code: "source_stale",
263 
      completed_at: DateTime.utc_now()
264 
    })
265 
    |> Repo.update()
302 
    Repo.transaction(fn ->
303 
      locked = Repo.get_for_update!(SemanticJob, job.id)
304
305 
      if locked.status == "running" and locked.attempts == job.attempts do
306 
        locked
307 
        |> SemanticJob.lifecycle_changeset(%{
308 
          status: "invalidated",
309 
          error_code: "source_stale",
310 
          completed_at: DateTime.utc_now()
311 
        })
312 
        |> update_or_rollback()
313 
      end
314 
    end)
266 315
267 316 
    {:ok, :invalidated}
268 317 
  end
lib/openagents/memory/semantic_worker.ex modified +49 -9

@@ -2,29 +2,69 @@ defmodule OpenAgents.Memory.SemanticWorker do

2 2 
  @moduledoc "Bounded asynchronous semantic outbox consumer; accepted turns never wait for it."
3 3 
  use GenServer
4 4
5 
  require Logger
6
5 7 
  alias OpenAgents.Memory.SemanticIndex
6 8
7 
  def start_link(options), do: GenServer.start_link(__MODULE__, options, name: __MODULE__)
9 
  def start_link(options) do
10 
    GenServer.start_link(__MODULE__, options, name: Keyword.get(options, :name, __MODULE__))
11 
  end
12
13 
  @doc false
14 
  def drain(worker \\ __MODULE__), do: GenServer.call(worker, :drain, 120_000)
8 15
9 16 
  @impl true
10 
  def init(_options) do
17 
  def init(options) do
11 18 
    config = Application.fetch_env!(:openagents, :semantic_index)
19 
    processor = Keyword.get(options, :processor, &process/1)
12 20
13 21 
    if Keyword.fetch!(config, :enabled) do
14 22 
      _manifest = SemanticIndex.ensure_manifest!(Map.new(config))
15 
      schedule(0)
23
24 
      if Keyword.get(options, :autostart, true), do: schedule(0)
16 25 
    end
17 26
18 
    {:ok, config}
27 
    {:ok, %{config: config, processor: processor}}
28 
  end
29
30 
  @impl true
31 
  def handle_call(:drain, _from, state) do
32 
    {result, state} = run_drain(state)
33 
    {:reply, result, state}
19 34 
  end
20 35
21 36 
  @impl true
22 
  def handle_info(:drain, config) do
23 
    provider = Keyword.fetch!(config, :provider)
24 
    _counts = SemanticIndex.process_all(provider, Keyword.fetch!(config, :batch_size))
25 
    schedule(Keyword.fetch!(config, :poll_interval_ms))
26 
    {:noreply, config}
37 
  def handle_info(:drain, state) do
38 
    {_result, state} = run_drain(state)
39 
    schedule(Keyword.fetch!(state.config, :poll_interval_ms))
40 
    {:noreply, state}
27 41 
  end
28 42
29 43 
  defp schedule(delay), do: Process.send_after(self(), :drain, delay)
44
45 
  defp run_drain(state) do
46 
    result =
47 
      try do
48 
        {:ok, state.processor.(state.config)}
49 
      rescue
50 
        _exception -> {:error, :semantic_drain_failed}
51 
      catch
52 
        _kind, _reason -> {:error, :semantic_drain_failed}
53 
      end
54
55 
    if result == {:error, :semantic_drain_failed} do
56 
      Logger.warning("semantic_drain_failed code=worker_exception")
57 
    end
58
59 
    {result, state}
60 
  end
61
62 
  defp process(config) do
63 
    SemanticIndex.process_all(
64 
      Keyword.fetch!(config, :provider),
65 
      Keyword.fetch!(config, :batch_size),
66 
      provider_timeout_ms: Keyword.get(config, :provider_timeout_ms, 15_000),
67 
      lease_ms: Keyword.get(config, :lease_ms, 30_000)
68 
    )
69 
  end
30 70 
end
lib/openagents/providers/open_ai.ex modified +31 -17

@@ -21,31 +21,45 @@ defmodule OpenAgents.Providers.OpenAI do

21 21
22 22 
  @impl true
23 23 
  def stream(%Request{} = request, on_event) when is_function(on_event, 1) do
24 
    with {:ok, api_key} <- fetch_api_key(),
25 
         {:ok, response} <- request(api_key, request) do
24 
    stream(request, on_event, [])
25 
  end
26
27 
  @doc false
28 
  def stream(%Request{} = request, on_event, options)
29 
      when is_function(on_event, 1) and is_list(options) do
30 
    with {:ok, api_key} <- fetch_api_key(options),
31 
         {:ok, response} <- request(api_key, request, options) do
26 32 
      consume_response(response, on_event)
27 33 
    end
28 34 
  end
29 35
30 
  defp fetch_api_key do
31 
    case OpenAgents.RuntimeConfig.fetch_secret(:openai_api_key) do
32 
      {:ok, key} -> {:ok, key}
33 
      {:error, :not_configured} -> {:error, :missing_api_key}
36 
  defp fetch_api_key(options) do
37 
    case Keyword.fetch(options, :api_key) do
38 
      {:ok, key} when is_binary(key) and byte_size(key) > 0 ->
39 
        {:ok, key}
40
41 
      _not_supplied ->
42 
        case OpenAgents.RuntimeConfig.fetch_secret(:openai_api_key) do
43 
          {:ok, key} -> {:ok, key}
44 
          {:error, :not_configured} -> {:error, :missing_api_key}
45 
        end
34 46 
    end
35 47 
  end
36 48
37 
  defp request(api_key, %Request{} = request) do
49 
  defp request(api_key, %Request{} = request, options) do
38 50 
    payload = request_payload(request)
39
40 
    case Req.post(@endpoint,
41 
           auth: {:bearer, api_key},
42 
           headers: [{"accept", "text/event-stream"}],
43 
           json: payload,
44 
           into: :self,
45 
           receive_timeout: 120_000,
46 
           retry: :transient,
47 
           max_retries: 2
48 
         ) do
51 
    request_options = Keyword.get(options, :request_options, [])
52
53 
    base_options = [
54 
      auth: {:bearer, api_key},
55 
      headers: [{"accept", "text/event-stream"}],
56 
      json: payload,
57 
      into: :self,
58 
      receive_timeout: 120_000,
59 
      retry: false
60 
    ]
61
62 
    case Req.post(@endpoint, Keyword.merge(base_options, request_options)) do
49 63 
      {:ok, response} ->
50 64 
        {:ok, response}
51 65
lib/openagents/runtime_config.ex modified +5 -1

@@ -726,7 +726,11 @@ defmodule OpenAgents.RuntimeConfig do

726 726 
      is_integer(keyword_value(settings, :batch_size)) and
727 727 
      keyword_value(settings, :batch_size) in 1..1_000 and
728 728 
      is_integer(keyword_value(settings, :poll_interval_ms)) and
729 
      keyword_value(settings, :poll_interval_ms) in 100..60_000
729 
      keyword_value(settings, :poll_interval_ms) in 100..60_000 and
730 
      is_integer(keyword_value(settings, :provider_timeout_ms)) and
731 
      keyword_value(settings, :provider_timeout_ms) in 100..120_000 and
732 
      is_integer(keyword_value(settings, :lease_ms)) and
733 
      keyword_value(settings, :lease_ms) in keyword_value(settings, :provider_timeout_ms)..300_000
730 734 
  end
731 735
732 736 
  defp valid_semantic_config?(_settings), do: false
lib/openagents/turn_recovery.ex modified +5 -2

@@ -8,8 +8,11 @@ defmodule OpenAgents.TurnRecovery do

8 8 
  end
9 9
10 10 
  @impl true
11 
  def init(_options) do
12 
    :ok = OpenAgents.Conversations.recover_interrupted_turns()
11 
  def init(options) do
12 
    recovery =
13 
      Keyword.get(options, :recovery, &OpenAgents.Conversations.recover_interrupted_turns/0)
14
15 
    :ok = recovery.()
13 16 
    {:ok, %{}}
14 17 
  end
15 18 
end
lib/openagents/voice/open_ai/call_client.ex modified +1 -2

@@ -69,8 +69,7 @@ defmodule OpenAgents.Voice.OpenAI.CallClient do

69 69 
        session: Jason.encode!(Config.session_payload(config))
70 70 
      ],
71 71 
      receive_timeout: 30_000,
72 
      retry: :transient,
73 
      max_retries: 1,
72 
      retry: false,
74 73 
      decode_body: false
75 74 
    ]
76 75
lib/openagents/voice_recovery.ex modified +2 -2

@@ -1,7 +1,7 @@

1 1 
defmodule OpenAgents.VoiceRecovery do
2 2 
  @moduledoc false
3 3
4 
  use GenServer
4 
  use GenServer, restart: :temporary
5 5
6 6 
  def start_link(options), do: GenServer.start_link(__MODULE__, options, name: __MODULE__)
7 7

@@ -9,6 +9,6 @@ defmodule OpenAgents.VoiceRecovery do

9 9 
  def init(options) do
10 10 
    recovery = Keyword.get(options, :recovery, &OpenAgents.Voice.recover_interrupted_sessions/0)
11 11 
    :ok = recovery.()
12 
    :ignore
12 
    {:ok, %{}}
13 13 
  end
14 14 
end
lib/openagents/work.ex modified +3 -3

@@ -683,9 +683,9 @@ defmodule OpenAgents.Work do

683 683 
  # computer — so the deep-work step summary is wrong and misleading for it. Give
684 684 
  # it an honest, kind-aware report that names the interruption and how to resume.
685 685 
  defp fallback_report(_repo, %Job{kind: "delegation"} = locked_job, status) do
686 
    params = locked_job.delegation || %{}
687 
    agent = params["agent_id"] || "the agent"
688 
    machine = params["machine_name"] || "the machine"
686 
    authority = locked_job.authority_snapshot || %{}
687 
    agent = authority["agent_id"] || "the agent"
688 
    machine = authority["machine_name"] || "the machine"
689 689
690 690 
    case status do
691 691 
      "interrupted" ->
lib/openagents/work/delegation_server.ex modified +24 -19

@@ -97,16 +97,18 @@ defmodule OpenAgents.Work.DelegationServer do

97 97
98 98 
  defp start_delegation(job, ra_gen, resume_id) do
99 99 
    params = job.delegation || %{}
100 
    authority = job.authority_snapshot || %{}
101 
    timeout_ms = timeout_ms(job)
100 102
101 103 
    payload =
102 104 
      %{
103 
        "agent_id" => params["agent_id"],
105 
        "agent_id" => authority["agent_id"],
104 106 
        "prompt" => params["prompt"],
105 
        "timeout_ms" => timeout_ms(params)
107 
        "timeout_ms" => timeout_ms
106 108 
      }
107 
      |> put_optional("cwd", params["cwd"])
109 
      |> put_optional("cwd", authority["cwd"])
108 110 
      |> put_optional("resume_session_id", resume_id || params["resume_session_id"])
109 
      |> attach_inference_grant(job, params)
111 
      |> attach_inference_grant(job)
110 112
111 113 
    # Checkpoint the ACP session id the moment the controller reports it — in
112 114 
    # Ra (cluster-wide, for node-loss handoff) AND in the durable job row

@@ -128,9 +130,9 @@ defmodule OpenAgents.Work.DelegationServer do

128 130 
        await_ms = if resume_id, do: 90_000, else: 0
129 131
130 132 
        Computer.request_agent(
131 
          params["machine_id"],
133 
          job.machine_id,
132 134 
          payload,
133 
          timeout_ms(params) + 15_000,
135 
          timeout_ms + 15_000,
134 136 
          on_session: on_session,
135 137 
          await_machine_ms: await_ms
136 138 
        )

@@ -157,12 +159,12 @@ defmodule OpenAgents.Work.DelegationServer do

157 159 
  # injects it into the probe process at spawn. Any other agent (which brings
158 160 
  # its own credential) gets nothing. A mint failure degrades to a
159 161 
  # grant-less delegation rather than blocking the work.
160 
  defp attach_inference_grant(payload, job, %{"agent_id" => "probe", "machine_id" => machine_id})
161 
       when is_binary(machine_id) do
162 
  defp attach_inference_grant(payload, %{authority_snapshot: %{"agent_id" => "probe"}} = job)
163 
       when is_binary(job.machine_id) do
162 164 
    case OpenAgents.Inference.mint(%{
163 165 
           owner_visitor_id: job.owner_visitor_id,
164 166 
           conversation_id: job.conversation_id,
165 
           machine_id: machine_id
167 
           machine_id: job.machine_id
166 168 
         }) do
167 169 
      {:ok, _grant, token} ->
168 170 
        payload

@@ -174,7 +176,7 @@ defmodule OpenAgents.Work.DelegationServer do

174 176 
    end
175 177 
  end
176 178
177 
  defp attach_inference_grant(payload, _job, _params), do: payload
179 
  defp attach_inference_grant(payload, _job), do: payload
178 180
179 181 
  defp summarize({:ok, %{"status" => "completed"} = payload}, job) do
180 182 
    {"completed", report_line(job, "completed", payload["output"], payload["detail"], payload)}

@@ -197,9 +199,9 @@ defmodule OpenAgents.Work.DelegationServer do

197 199 
  end
198 200
199 201 
  defp report_line(job, status, output, detail, payload) do
200 
    params = job.delegation || %{}
201 
    agent = params["agent_id"] || "agent"
202 
    machine = params["machine_name"] || "the machine"
202 
    authority = job.authority_snapshot || %{}
203 
    agent = authority["agent_id"] || "agent"
204 
    machine = authority["machine_name"] || "the machine"
203 205 
    header = "Delegation to #{agent} on #{machine} — #{human_status(status)}."
204 206
205 207 
    body =

@@ -282,8 +284,11 @@ defmodule OpenAgents.Work.DelegationServer do

282 284 
    end
283 285 
  end
284 286
285 
  defp timeout_ms(%{"timeout_ms" => value}) when is_integer(value) and value > 0, do: value
286 
  defp timeout_ms(_params), do: 3_600_000
287 
  defp timeout_ms(%{budget_snapshot: %{"wall_clock_ms" => value}})
288 
       when is_integer(value) and value > 0,
289 
       do: value
290
291 
  defp timeout_ms(_job), do: 3_600_000
287 292
288 293 
  defp put_optional(payload, _key, value) when value in [nil, ""], do: payload
289 294 
  defp put_optional(payload, key, value), do: Map.put(payload, key, value)

@@ -300,7 +305,7 @@ defmodule OpenAgents.Work.DelegationServer do

300 305 
  defp write_incident(job, result, status) do
301 306 
    code = incident_code(status)
302 307 
    payload = incident_payload(result)
303 
    params = job.delegation || %{}
308 
    authority = job.authority_snapshot || %{}
304 309 
    owner_user_id = incident_owner_user_id(job)
305 310
306 311 
    Incidents.report(%{

@@ -313,9 +318,9 @@ defmodule OpenAgents.Work.DelegationServer do

313 318 
      code: code,
314 319 
      summary: "Delegation #{human_status(status)}: #{code}",
315 320 
      context: %{
316 
        "cwd" => params["cwd"] || "",
317 
        "agent_id" => params["agent_id"] || "",
318 
        "machine_id" => params["machine_id"] || "",
321 
        "cwd" => authority["cwd"] || "",
322 
        "agent_id" => authority["agent_id"] || "",
323 
        "machine_id" => job.machine_id || "",
319 324 
        "duration_ms" => payload["duration_ms"] || 0,
320 325 
        "truncated" => payload["truncated"] || false,
321 326 
        "session_id" => payload["session_id"] || ""
lib/openagents/work/job.ex modified +78

@@ -16,6 +16,7 @@ defmodule OpenAgents.Work.Job do

16 16 
  @terminal_statuses ~w(completed failed interrupted budget_exhausted cancelled)
17 17 
  @surfaces ~w(text voice)
18 18 
  @kinds ~w(deep_work delegation coding)
19 
  @machine_tiers ~w(probe curated shell)
19 20 
  @maximum_goal_bytes 2_000
20 21 
  @maximum_context_hint_bytes 2_000
21 22 
  @maximum_report_bytes 8_000

@@ -27,12 +28,15 @@ defmodule OpenAgents.Work.Job do

27 28 
  schema "work_jobs" do
28 29 
    belongs_to :conversation, OpenAgents.Conversations.Conversation
29 30 
    belongs_to :owner_visitor, OpenAgents.Conversations.Visitor
31 
    belongs_to :machine, OpenAgents.Machines.Machine
30 32 
    field :surface, :string
31 33 
    field :goal, :string
32 34 
    field :context_hint, :string
33 35 
    field :requesting_tool_step_ref, :string
34 36 
    field :kind, :string, default: "deep_work"
35 37 
    field :delegation, :map
38 
    field :authority_snapshot, :map
39 
    field :budget_snapshot, :map
36 40 
    field :status, :string, default: "queued"
37 41 
    field :report, :string
38 42 
    field :error_code, :string

@@ -72,14 +76,19 @@ defmodule OpenAgents.Work.Job do

72 76 
    ])
73 77 
    |> put_change(:conversation_id, Map.fetch!(attributes, :conversation_id))
74 78 
    |> put_change(:owner_visitor_id, Map.fetch!(attributes, :owner_visitor_id))
79 
    |> put_optional_identity(:machine_id, attributes)
80 
    |> put_optional_identity(:authority_snapshot, attributes)
81 
    |> put_optional_identity(:budget_snapshot, attributes)
75 82 
    |> validate_required([:conversation_id, :owner_visitor_id, :surface, :goal])
76 83 
    |> validate_inclusion(:surface, @surfaces)
77 84 
    |> validate_inclusion(:kind, @kinds)
78 85 
    |> validate_byte_length(:goal, @maximum_goal_bytes)
79 86 
    |> validate_byte_length(:context_hint, @maximum_context_hint_bytes)
80 87 
    |> validate_length(:requesting_tool_step_ref, max: 256)
88 
    |> validate_delegation_identity()
81 89 
    |> foreign_key_constraint(:conversation_id)
82 90 
    |> foreign_key_constraint(:owner_visitor_id)
91 
    |> foreign_key_constraint(:machine_id)
83 92 
  end
84 93
85 94 
  @doc "Moves the job through its running lifecycle without touching identity."

@@ -118,6 +127,75 @@ defmodule OpenAgents.Work.Job do

118 127 
    end
119 128 
  end
120 129
130 
  defp put_optional_identity(changeset, key, attributes) do
131 
    case Map.fetch(attributes, key) do
132 
      {:ok, value} -> put_change(changeset, key, value)
133 
      :error -> changeset
134 
    end
135 
  end
136
137 
  defp validate_delegation_identity(changeset) do
138 
    if get_field(changeset, :kind) == "delegation" do
139 
      changeset
140 
      |> validate_required([:machine_id, :authority_snapshot, :budget_snapshot, :delegation])
141 
      |> validate_snapshot(:authority_snapshot, 32_768)
142 
      |> validate_snapshot(:budget_snapshot, 4_096)
143 
      |> validate_delegation_snapshot_match()
144 
    else
145 
      changeset
146 
    end
147 
  end
148
149 
  defp validate_delegation_snapshot_match(changeset) do
150 
    machine_id = get_field(changeset, :machine_id)
151 
    delegation = get_field(changeset, :delegation)
152 
    authority = get_field(changeset, :authority_snapshot)
153 
    budget = get_field(changeset, :budget_snapshot)
154
155 
    valid? =
156 
      is_map(delegation) and is_map(authority) and is_map(budget) and
157 
        delegation["machine_id"] == machine_id and
158 
        delegation["agent_id"] == authority["agent_id"] and
159 
        delegation["cwd"] == authority["cwd"] and
160 
        delegation["machine_name"] == authority["machine_name"] and
161 
        delegation["timeout_ms"] == budget["wall_clock_ms"] and
162 
        bounded_string?(delegation["prompt"], budget["maximum_prompt_bytes"]) and
163 
        bounded_string?(authority["agent_id"], 64) and
164 
        bounded_string?(authority["cwd"], 500) and
165 
        bounded_string?(authority["machine_name"], 256) and
166 
        valid_roots?(authority["roots"]) and
167 
        authority["machine_tier"] in @machine_tiers and
168 
        budget["maximum_report_bytes"] == @maximum_report_bytes and
169 
        budget["wall_clock_ms"] in 1..3_600_000
170
171 
    if valid?,
172 
      do: changeset,
173 
      else: add_error(changeset, :delegation, "does not match the admitted execution snapshot")
174 
  end
175
176 
  defp bounded_string?(value, maximum)
177 
       when is_binary(value) and is_integer(maximum) and maximum > 0,
178 
       do: value != "" and byte_size(value) <= maximum
179
180 
  defp bounded_string?(_value, _maximum), do: false
181
182 
  defp valid_roots?(roots) when is_list(roots) and roots != [],
183 
    do: Enum.all?(roots, &bounded_string?(&1, 500))
184
185 
  defp valid_roots?(_roots), do: false
186
187 
  defp validate_snapshot(changeset, field, maximum_bytes) do
188 
    case get_field(changeset, field) do
189 
      value when is_map(value) ->
190 
        if byte_size(Jason.encode!(value)) <= maximum_bytes,
191 
          do: changeset,
192 
          else: add_error(changeset, field, "exceeds #{maximum_bytes} bytes")
193
194 
      _invalid ->
195 
        changeset
196 
    end
197 
  end
198
121 199 
  defp validate_byte_length(changeset, field, maximum) do
122 200 
    case get_field(changeset, field) do
123 201 
      nil ->
lib/openagents/work_recovery.ex modified +3 -2

@@ -8,8 +8,9 @@ defmodule OpenAgents.WorkRecovery do

8 8 
  end
9 9
10 10 
  @impl true
11 
  def init(_options) do
12 
    :ok = OpenAgents.Work.recover_interrupted_jobs()
11 
  def init(options) do
12 
    recovery = Keyword.get(options, :recovery, &OpenAgents.Work.recover_interrupted_jobs/0)
13 
    :ok = recovery.()
13 14 
    {:ok, %{}}
14 15 
  end
15 16 
end
priv/repo/migrations/20260820085203_harden_async_runtime_boundaries.exs added +194

@@ -0,0 +1,194 @@

1 
defmodule OpenAgents.Repo.Migrations.HardenAsyncRuntimeBoundaries do
2 
  use Ecto.Migration
3
4 
  def up do
5 
    create constraint(:messages, :messages_content_hard_bound,
6 
             check: "octet_length(content) <= 1048576"
7 
           )
8
9 
    create unique_index(:messages, [:id, :conversation_id],
10 
             name: :messages_id_conversation_id_index
11 
           )
12
13 
    execute("""
14 
    ALTER TABLE semantic_embedding_jobs
15 
    ADD CONSTRAINT semantic_jobs_message_scope_fk
16 
    FOREIGN KEY (message_id, conversation_id)
17 
    REFERENCES messages(id, conversation_id)
18 
    ON DELETE CASCADE
19 
    """)
20
21 
    execute("""
22 
    ALTER TABLE message_semantic_embeddings
23 
    ADD CONSTRAINT semantic_embeddings_message_scope_fk
24 
    FOREIGN KEY (message_id, conversation_id)
25 
    REFERENCES messages(id, conversation_id)
26 
    ON DELETE CASCADE
27 
    """)
28
29 
    alter table(:work_jobs) do
30 
      add :machine_id, references(:machines, type: :binary_id, on_delete: :restrict)
31 
      add :authority_snapshot, :map
32 
      add :budget_snapshot, :map
33 
    end
34
35 
    create index(:work_jobs, [:machine_id, :inserted_at])
36
37 
    execute("""
38 
    UPDATE work_jobs AS job
39 
    SET machine_id = machine.id,
40 
        authority_snapshot = jsonb_build_object(
41 
          'machine_tier', machine.tier,
42 
          'roots', machine.roots,
43 
          'cwd', COALESCE(job.delegation->>'cwd', ''),
44 
          'agent_id', COALESCE(job.delegation->>'agent_id', ''),
45 
          'machine_name', COALESCE(job.delegation->>'machine_name', machine.name)
46 
        ),
47 
        budget_snapshot = jsonb_build_object(
48 
          'wall_clock_ms', CASE
49 
            WHEN job.delegation->>'timeout_ms' ~ '^[0-9]+$'
50 
              THEN (job.delegation->>'timeout_ms')::integer
51 
            ELSE 3600000
52 
          END,
53 
          'maximum_prompt_bytes', 8000,
54 
          'maximum_report_bytes', 8000
55 
        )
56 
    FROM machines AS machine
57 
    WHERE job.kind = 'delegation'
58 
      AND job.delegation->>'machine_id' = machine.id::text
59 
    """)
60
61 
    create constraint(:work_jobs, :work_jobs_delegation_identity,
62 
             check:
63 
               "kind <> 'delegation' OR (machine_id IS NOT NULL AND jsonb_typeof(delegation) = 'object' AND jsonb_typeof(authority_snapshot) = 'object' AND jsonb_typeof(budget_snapshot) = 'object' AND octet_length(authority_snapshot::text) <= 32768 AND octet_length(budget_snapshot::text) <= 4096 AND jsonb_typeof(authority_snapshot->'roots') = 'array' AND jsonb_array_length(authority_snapshot->'roots') > 0 AND authority_snapshot->>'machine_tier' IN ('probe', 'curated', 'shell') AND jsonb_typeof(authority_snapshot->'agent_id') = 'string' AND octet_length(authority_snapshot->>'agent_id') BETWEEN 1 AND 64 AND jsonb_typeof(authority_snapshot->'cwd') = 'string' AND octet_length(authority_snapshot->>'cwd') BETWEEN 1 AND 500 AND jsonb_typeof(authority_snapshot->'machine_name') = 'string' AND octet_length(authority_snapshot->>'machine_name') BETWEEN 1 AND 256 AND jsonb_typeof(delegation->'prompt') = 'string' AND octet_length(delegation->>'prompt') BETWEEN 1 AND 8000 AND jsonb_typeof(delegation->'timeout_ms') = 'number' AND jsonb_typeof(budget_snapshot->'wall_clock_ms') = 'number' AND (budget_snapshot->>'wall_clock_ms')::numeric BETWEEN 1 AND 3600000 AND jsonb_typeof(budget_snapshot->'maximum_prompt_bytes') = 'number' AND (budget_snapshot->>'maximum_prompt_bytes')::numeric BETWEEN 1 AND 8000 AND jsonb_typeof(budget_snapshot->'maximum_report_bytes') = 'number' AND (budget_snapshot->>'maximum_report_bytes')::numeric = 8000 AND delegation->>'machine_id' = machine_id::text AND delegation->>'agent_id' = authority_snapshot->>'agent_id' AND delegation->>'cwd' = authority_snapshot->>'cwd' AND delegation->>'machine_name' = authority_snapshot->>'machine_name' AND (delegation->>'timeout_ms')::numeric = (budget_snapshot->>'wall_clock_ms')::numeric AND octet_length(delegation->>'prompt') <= (budget_snapshot->>'maximum_prompt_bytes')::numeric)"
64 
           )
65
66 
    execute("""
67 
    CREATE FUNCTION enforce_work_job_scope()
68 
    RETURNS trigger AS $$
69 
    BEGIN
70 
      IF NOT EXISTS (
71 
        SELECT 1 FROM conversations
72 
        WHERE id = NEW.conversation_id AND visitor_id = NEW.owner_visitor_id
73 
      ) THEN
74 
        RAISE EXCEPTION 'work job conversation owner mismatch';
75 
      END IF;
76
77 
      IF NEW.kind = 'delegation' AND NOT EXISTS (
78 
        SELECT 1
79 
        FROM machines AS machine
80 
        JOIN visitors AS visitor ON visitor.id = NEW.owner_visitor_id
81 
        WHERE machine.id = NEW.machine_id
82 
          AND visitor.user_id IS NOT NULL
83 
          AND machine.user_id = visitor.user_id
84 
      ) THEN
85 
        RAISE EXCEPTION 'work job machine owner mismatch';
86 
      END IF;
87
88 
      IF TG_OP = 'INSERT' AND NEW.kind = 'delegation' AND NOT EXISTS (
89 
        SELECT 1
90 
        FROM machines AS machine
91 
        WHERE machine.id = NEW.machine_id
92 
          AND NEW.authority_snapshot->>'machine_tier' = machine.tier
93 
          AND NEW.authority_snapshot->'roots' = to_jsonb(machine.roots)
94 
          AND NEW.authority_snapshot->>'machine_name' = machine.name
95 
      ) THEN
96 
        RAISE EXCEPTION 'work job machine authority snapshot mismatch';
97 
      END IF;
98
99 
      RETURN NEW;
100 
    END;
101 
    $$ LANGUAGE plpgsql;
102 
    """)
103
104 
    execute("""
105 
    CREATE TRIGGER work_jobs_enforce_scope
106 
    BEFORE INSERT OR UPDATE ON work_jobs
107 
    FOR EACH ROW
108 
    EXECUTE FUNCTION enforce_work_job_scope();
109 
    """)
110
111 
    execute(identity_function(true))
112 
  end
113
114 
  def down do
115 
    execute(identity_function(false))
116 
    execute("DROP TRIGGER IF EXISTS work_jobs_enforce_scope ON work_jobs")
117 
    execute("DROP FUNCTION IF EXISTS enforce_work_job_scope()")
118 
    drop constraint(:work_jobs, :work_jobs_delegation_identity)
119 
    drop index(:work_jobs, [:machine_id, :inserted_at])
120
121 
    alter table(:work_jobs) do
122 
      remove :budget_snapshot
123 
      remove :authority_snapshot
124 
      remove :machine_id
125 
    end
126
127 
    execute(
128 
      "ALTER TABLE message_semantic_embeddings DROP CONSTRAINT IF EXISTS semantic_embeddings_message_scope_fk"
129 
    )
130
131 
    execute(
132 
      "ALTER TABLE semantic_embedding_jobs DROP CONSTRAINT IF EXISTS semantic_jobs_message_scope_fk"
133 
    )
134
135 
    drop_if_exists index(:messages, [:id, :conversation_id],
136 
                     name: :messages_id_conversation_id_index
137 
                   )
138
139 
    drop constraint(:messages, :messages_content_hard_bound)
140 
  end
141
142 
  defp identity_function(include_delegation_identity?) do
143 
    extra_old =
144 
      if include_delegation_identity?,
145 
        do:
146 
          ", OLD.kind, CASE WHEN OLD.kind = 'delegation' THEN OLD.delegation - 'resume_session_id' ELSE NULL END, OLD.machine_id, OLD.authority_snapshot, OLD.budget_snapshot",
147 
        else: ""
148
149 
    extra_new =
150 
      if include_delegation_identity?,
151 
        do:
152 
          ", NEW.kind, CASE WHEN NEW.kind = 'delegation' THEN NEW.delegation - 'resume_session_id' ELSE NULL END, NEW.machine_id, NEW.authority_snapshot, NEW.budget_snapshot",
153 
        else: ""
154
155 
    """
156 
    CREATE OR REPLACE FUNCTION enforce_work_job_transition()
157 
    RETURNS trigger AS $$
158 
    BEGIN
159 
      IF ROW(
160 
        OLD.conversation_id, OLD.owner_visitor_id, OLD.surface, OLD.goal,
161 
        OLD.context_hint, OLD.requesting_tool_step_ref#{extra_old}
162 
      ) IS DISTINCT FROM ROW(
163 
        NEW.conversation_id, NEW.owner_visitor_id, NEW.surface, NEW.goal,
164 
        NEW.context_hint, NEW.requesting_tool_step_ref#{extra_new}
165 
      ) THEN
166 
        RAISE EXCEPTION 'work job identity is immutable';
167 
      END IF;
168
169 
      IF OLD.status = 'queued' AND NEW.status NOT IN (
170 
        'queued', 'running', 'failed', 'interrupted', 'cancelled'
171 
      ) THEN
172 
        RAISE EXCEPTION 'invalid queued work job transition';
173 
      END IF;
174
175 
      IF OLD.status = 'running' AND NEW.status NOT IN (
176 
        'running', 'completed', 'failed', 'interrupted', 'budget_exhausted', 'cancelled'
177 
      ) THEN
178 
        RAISE EXCEPTION 'invalid running work job transition';
179 
      END IF;
180
181 
      IF OLD.status NOT IN ('queued', 'running') AND ROW(
182 
        OLD.status, OLD.report, OLD.error_code, OLD.usage, OLD.completed_at
183 
      ) IS DISTINCT FROM ROW(
184 
        NEW.status, NEW.report, NEW.error_code, NEW.usage, NEW.completed_at
185 
      ) THEN
186 
        RAISE EXCEPTION 'terminal work job is immutable';
187 
      END IF;
188
189 
      RETURN NEW;
190 
    END;
191 
    $$ LANGUAGE plpgsql;
192 
    """
193 
  end
194 
end
test/openagents/conversations_test.exs modified +16

@@ -95,6 +95,22 @@ defmodule OpenAgents.ConversationsTest do

95 95 
             Conversations.create_turn(conversation, String.duplicate("x", 8_001))
96 96 
  end
97 97
98 
  test "assistant streaming stops before the accumulated message exceeds its byte budget" do
99 
    assert {:ok, conversation} = Conversations.ensure_conversation("assistant-bound-browser")
100 
    assert {:ok, records} = Conversations.create_turn(conversation, "Keep the reply bounded")
101 
    maximum = Application.fetch_env!(:openagents, :maximum_message_bytes)
102
103 
    assert {:ok, message} =
104 
             Conversations.append_assistant_delta(records.turn, String.duplicate("a", maximum))
105
106 
    assert byte_size(message.content) == maximum
107
108 
    assert {:error, :assistant_message_limit_reached} =
109 
             Conversations.append_assistant_delta(records.turn, "b")
110
111 
    assert byte_size(Repo.get!(Message, records.assistant_message.id).content) == maximum
112 
  end
113
98 114 
  test "startup recovery makes interrupted turns explicitly failed" do
99 115 
    assert {:ok, conversation} = Conversations.ensure_conversation("recovery-browser")
100 116 
    assert {:ok, records} = Conversations.create_turn(conversation, "Do not leave this pending")
test/openagents/memory/semantic_worker_test.exs added +167

@@ -0,0 +1,167 @@

1 
defmodule OpenAgents.Memory.SemanticWorkerTest do
2 
  use OpenAgents.DataCase, async: false
3
4 
  alias OpenAgents.Conversations
5 
  alias OpenAgents.Memory.{SemanticIndex, SemanticJob, SemanticWorker}
6
7 
  @provider OpenAgents.Memory.SemanticWorkerTestProvider
8
9 
  setup do
10 
    original_config = Application.fetch_env!(:openagents, :semantic_index)
11 
    original_mode = Application.get_env(:openagents, :semantic_worker_test_mode)
12
13 
    Application.put_env(:openagents, :semantic_index,
14 
      enabled: true,
15 
      provider: @provider,
16 
      model_id: "semantic-worker-test",
17 
      model_version: "v1",
18 
      dimensions: 64,
19 
      batch_size: 20,
20 
      poll_interval_ms: 60_000,
21 
      provider_timeout_ms: 5_000,
22 
      lease_ms: 10_000
23 
    )
24
25 
    on_exit(fn ->
26 
      Application.put_env(:openagents, :semantic_index, original_config)
27
28 
      if is_nil(original_mode),
29 
        do: Application.delete_env(:openagents, :semantic_worker_test_mode),
30 
        else: Application.put_env(:openagents, :semantic_worker_test_mode, original_mode)
31 
    end)
32
33 
    :ok
34 
  end
35
36 
  test "provider failure becomes a bounded durable failure without killing the worker" do
37 
    Application.put_env(:openagents, :semantic_worker_test_mode, :failure)
38 
    worker = start_worker(:semantic_failure_worker)
39 
    assert {:ok, _conversation} = Conversations.ensure_conversation("semantic-worker-failure")
40
41 
    assert {:ok, %{completed: 0, failed: 1, invalidated: 0}} =
42 
             SemanticWorker.drain(worker)
43
44 
    assert %SemanticJob{status: "failed", error_code: "semantic_provider_offline", attempts: 1} =
45 
             Repo.one!(SemanticJob)
46
47 
    _state = :sys.get_state(worker)
48 
  end
49
50 
  test "a drain exception is contained so a later database pass can succeed" do
51 
    Application.put_env(:openagents, :semantic_worker_test_mode, :success)
52 
    attempts = start_supervised!({Agent, fn -> 0 end})
53
54 
    processor = fn _config ->
55 
      case Agent.get_and_update(attempts, &{&1, &1 + 1}) do
56 
        0 -> raise "database unavailable"
57 
        _later -> %{completed: 0, failed: 0, invalidated: 0}
58 
      end
59 
    end
60
61 
    worker = start_worker(:semantic_database_worker, processor: processor)
62
63 
    assert {:error, :semantic_drain_failed} = SemanticWorker.drain(worker)
64 
    assert {:ok, %{completed: 0, failed: 0, invalidated: 0}} = SemanticWorker.drain(worker)
65 
    _state = :sys.get_state(worker)
66 
  end
67
68 
  test "an expired running lease is reclaimed after the worker process dies" do
69 
    Application.put_env(:openagents, :semantic_worker_test_mode, {:block, self()})
70 
    assert {:ok, _conversation} = Conversations.ensure_conversation("semantic-worker-reclaim")
71 
    first = start_worker(:semantic_crash_worker)
72
73 
    {caller, caller_ref} =
74 
      spawn_monitor(fn ->
75 
        SemanticWorker.drain(first)
76 
      end)
77
78 
    assert_receive {:semantic_provider_started, provider_task}
79 
    assert %SemanticJob{status: "running", attempts: 1} = Repo.one!(SemanticJob)
80
81 
    Repo.update_all(SemanticJob,
82 
      set: [available_at: DateTime.add(DateTime.utc_now(), -1, :second)]
83 
    )
84
85 
    provider_ref = Process.monitor(provider_task)
86 
    worker_ref = Process.monitor(first)
87 
    Process.exit(first, :kill)
88 
    assert_receive {:DOWN, ^worker_ref, :process, ^first, :killed}
89 
    assert_receive {:DOWN, ^caller_ref, :process, ^caller, _reason}
90 
    send(provider_task, :release_semantic_provider)
91 
    assert_receive {:DOWN, ^provider_ref, :process, ^provider_task, _reason}, 1_000
92
93 
    Application.put_env(:openagents, :semantic_worker_test_mode, :success)
94 
    second = start_worker(:semantic_replacement_worker)
95
96 
    assert {:ok, %{completed: 1, failed: 0, invalidated: 0}} =
97 
             SemanticWorker.drain(second)
98
99 
    assert %SemanticJob{status: "completed", attempts: 2, error_code: nil} =
100 
             Repo.one!(SemanticJob)
101
102 
    assert embedding_count() == 1
103 
  end
104
105 
  test "a reclaimed attempt fences the first provider result" do
106 
    Application.put_env(:openagents, :semantic_worker_test_mode, {:block, self()})
107 
    config = Application.fetch_env!(:openagents, :semantic_index) |> Map.new()
108 
    _manifest = SemanticIndex.ensure_manifest!(config)
109 
    assert {:ok, _conversation} = Conversations.ensure_conversation("semantic-attempt-fence")
110 
    observer = self()
111
112 
    first =
113 
      start_supervised!(
114 
        {Task,
115 
         fn ->
116 
           result =
117 
             SemanticIndex.process_next(@provider,
118 
               provider_timeout_ms: 30_000,
119 
               lease_ms: 30_000
120 
             )
121
122 
           send(observer, {:first_semantic_result, result})
123 
         end}
124 
      )
125
126 
    assert_receive {:semantic_provider_started, provider_task}
127 
    assert %SemanticJob{status: "running", attempts: 1} = Repo.one!(SemanticJob)
128
129 
    Repo.update_all(SemanticJob,
130 
      set: [available_at: DateTime.add(DateTime.utc_now(), -1, :second)]
131 
    )
132
133 
    Application.put_env(:openagents, :semantic_worker_test_mode, :success)
134
135 
    assert {:ok, :completed} =
136 
             SemanticIndex.process_next(@provider,
137 
               provider_timeout_ms: 5_000,
138 
               lease_ms: 10_000
139 
             )
140
141 
    first_ref = Process.monitor(first)
142 
    send(provider_task, :release_semantic_provider)
143 
    assert_receive {:first_semantic_result, {:ok, :invalidated}}
144 
    assert_receive {:DOWN, ^first_ref, :process, ^first, :normal}
145
146 
    assert %SemanticJob{status: "completed", attempts: 2} = Repo.one!(SemanticJob)
147 
    assert embedding_count() == 1
148 
  end
149
150 
  defp start_worker(id, options \\ []) do
151 
    name = Module.concat(__MODULE__, id)
152
153 
    spec =
154 
      Supervisor.child_spec(
155 
        {SemanticWorker, Keyword.merge([name: name, autostart: false], options)},
156 
        id: id,
157 
        restart: :temporary
158 
      )
159
160 
    start_supervised!(spec)
161 
  end
162
163 
  defp embedding_count do
164 
    %{rows: [[count]]} = Repo.query!("SELECT count(*) FROM message_semantic_embeddings")
165 
    count
166 
  end
167 
end
test/openagents/providers/provider_contract_test.exs modified +19 -1

@@ -1,7 +1,9 @@

1 1 
defmodule OpenAgents.Providers.ProviderContractTest do
2 2 
  use ExUnit.Case, async: true
3 3
4 
  alias OpenAgents.Providers.{OpenAI, Test}
4 
  alias OpenAgents.Providers.{OpenAI, Request, Test}
5
6 
  setup {Req.Test, :verify_on_exit!}
5 7
6 8 
  test "providers expose stable IDs and finite provider-neutral capabilities" do
7 9 
    for provider <- [OpenAI, Test] do

@@ -16,6 +18,22 @@ defmodule OpenAgents.Providers.ProviderContractTest do

16 18 
    assert __MODULE__.TextOnlyProvider.capabilities() == [:text]
17 19 
  end
18 20
21 
  test "response creation does not retry a failed POST without an idempotency contract" do
22 
    Req.Test.expect(__MODULE__, fn conn -> Plug.Conn.send_resp(conn, 503, "unavailable") end)
23
24 
    request = %Request{
25 
      model_id: "test-model",
26 
      instructions: "Bounded test instructions",
27 
      input: [%{role: "user", content: "Hello"}]
28 
    }
29
30 
    assert {:error, {:http_status, 503}} =
31 
             OpenAI.stream(request, fn _event -> :ok end,
32 
               api_key: "test-secret",
33 
               request_options: [plug: {Req.Test, __MODULE__}]
34 
             )
35 
  end
36
19 37 
  defmodule TextOnlyProvider do
20 38 
    @behaviour OpenAgents.Providers.Provider
21 39
test/openagents/semantic_recall_test.exs modified +29

@@ -166,6 +166,35 @@ defmodule OpenAgents.SemanticRecallTest do

166 166 
    assert hd(fallback.matches).source_ref == "message:#{source.id}"
167 167 
  end
168 168
169 
  test "the database prevents moving a semantic job across conversation scope" do
170 
    first = conversation("semantic-job-scope-first")
171 
    second = conversation("semantic-job-scope-second")
172 
    source = message(first, "Keep this job in its source conversation.")
173
174 
    assert_raise Postgrex.Error, fn ->
175 
      Repo.update_all(
176 
        from(job in SemanticJob, where: job.message_id == ^source.id),
177 
        set: [conversation_id: second.id]
178 
      )
179 
    end
180 
  end
181
182 
  test "the database prevents moving an embedding across conversation scope" do
183 
    first = conversation("semantic-vector-scope-first")
184 
    second = conversation("semantic-vector-scope-second")
185 
    source = message(first, "Keep this vector in its source conversation.")
186
187 
    assert %{failed: 0, invalidated: 0} =
188 
             SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
189
190 
    assert_raise Postgrex.Error, fn ->
191 
      Repo.query!(
192 
        "UPDATE message_semantic_embeddings SET conversation_id=$1::text::uuid WHERE message_id=$2::text::uuid",
193 
        [second.id, source.id]
194 
      )
195 
    end
196 
  end
197
169 198 
  test "committed hybrid comparison improves synonym recall without weakening the lexical baseline",
170 199 
       %{original_semantic_config: original} do
171 200 
    conversation = conversation("semantic-release-eval")
test/openagents/tools/incident_lookup_test.exs modified +38 -2

@@ -3,7 +3,7 @@ defmodule OpenAgents.Tools.IncidentLookupTest do

3 3
4 4 
  alias OpenAgents.Incidents
5 5 
  alias OpenAgents.Tools.{ExecutionContext, Registry, Runner}
6 
  alias OpenAgents.{Accounts, Conversations, Repo}
6 
  alias OpenAgents.{Accounts, Conversations, Machines, Repo}
7 7
8 8 
  setup do
9 9 
    assert {:ok, snapshot} = Registry.build([OpenAgents.Tools.IncidentLookup])

@@ -63,6 +63,7 @@ defmodule OpenAgents.Tools.IncidentLookupTest do

63 63
64 64 
  test "a newer job report is primary over a stale incident", %{snapshot: snapshot} do
65 65 
    scope = owner_scope("incident-tool-stale")
66 
    machine = machine_for(scope.user)
66 67
67 68 
    {:ok, _} =
68 69 
      Incidents.record(%{

@@ -79,9 +80,30 @@ defmodule OpenAgents.Tools.IncidentLookupTest do

79 80 
      OpenAgents.Work.create_job(%{
80 81 
        conversation_id: scope.conversation.id,
81 82 
        owner_visitor_id: scope.owner.id,
83 
        machine_id: machine.id,
82 84 
        surface: "text",
83 85 
        kind: "delegation",
84 
        goal: "Delegate to claude: input-bar refactor"
86 
        goal: "Delegate to claude: input-bar refactor",
87 
        delegation: %{
88 
          "agent_id" => "claude",
89 
          "machine_id" => machine.id,
90 
          "machine_name" => machine.name,
91 
          "prompt" => "input-bar refactor",
92 
          "cwd" => "/tmp/openagents-incidents",
93 
          "timeout_ms" => 3_600_000
94 
        },
95 
        authority_snapshot: %{
96 
          "machine_tier" => machine.tier,
97 
          "roots" => machine.roots,
98 
          "cwd" => "/tmp/openagents-incidents",
99 
          "agent_id" => "claude",
100 
          "machine_name" => machine.name
101 
        },
102 
        budget_snapshot: %{
103 
          "wall_clock_ms" => 3_600_000,
104 
          "maximum_prompt_bytes" => 8_000,
105 
          "maximum_report_bytes" => 8_000
106 
        }
85 107 
      })
86 108
87 109 
    {:ok, running} = OpenAgents.Work.mark_job_running(job, %{})

@@ -176,4 +198,18 @@ defmodule OpenAgents.Tools.IncidentLookupTest do

176 198 
    owner = Repo.get!(OpenAgents.Conversations.Visitor, conversation.visitor_id)
177 199 
    %{user: user, owner: owner, conversation: conversation}
178 200 
  end
201
202 
  defp machine_for(user) do
203 
    {:ok, %{code: code}} =
204 
      Machines.start_pairing(%{
205 
        "name" => "incident-test-machine",
206 
        "tier" => "curated",
207 
        "platform" => "linux-x64",
208 
        "agent_version" => "0.1.0",
209 
        "roots" => ["/tmp/openagents-incidents"]
210 
      })
211
212 
    {:ok, machine} = Machines.approve_pairing(user, code)
213 
    machine
214 
  end
179 215 
end
test/openagents/turn_provenance_test.exs modified +32

@@ -146,6 +146,38 @@ defmodule OpenAgents.TurnProvenanceTest do

146 146 
             Conversations.list_provider_steps(recovered_receipt)
147 147 
  end
148 148
149 
  test "TurnRecovery finalizes durable evidence after the live server dies mid-stream" do
150 
    Application.put_env(:openagents, :test_provider_observer, self())
151 
    on_exit(fn -> Application.delete_env(:openagents, :test_provider_observer) end)
152
153 
    assert {:ok, conversation} = Conversations.ensure_conversation("turn-worker-recovery")
154 
    assert {:ok, records} = Conversations.create_turn(conversation, "[observe-request]")
155 
    assert {:ok, turn_server} = Turns.start(records.turn.id)
156 
    assert_receive {:provider_request, provider_task, _request}
157
158 
    server_ref = Process.monitor(turn_server)
159 
    Process.exit(turn_server, :kill)
160 
    assert_receive {:DOWN, ^server_ref, :process, ^turn_server, :killed}
161 
    provider_ref = Process.monitor(provider_task)
162 
    send(provider_task, :continue_provider)
163 
    assert_receive {:DOWN, ^provider_ref, :process, ^provider_task, _reason}, 1_000
164
165 
    recovery = start_supervised!({OpenAgents.TurnRecovery, []})
166 
    _state = :sys.get_state(recovery)
167
168 
    recovered_turn = Conversations.get_turn!(records.turn.id)
169 
    recovered_message = Repo.get!(OpenAgents.Conversations.Message, records.assistant_message.id)
170 
    {:ok, recovered_receipt} = Conversations.get_turn_receipt(recovered_turn)
171
172 
    assert recovered_turn.status == "failed"
173 
    assert recovered_turn.error_code == "runtime_restarted"
174 
    assert recovered_message.status == "failed"
175 
    assert recovered_receipt.status == "interrupted"
176
177 
    assert [%ProviderStep{status: "interrupted", error_code: "runtime_restarted"}] =
178 
             Conversations.list_provider_steps(recovered_receipt)
179 
  end
180
149 181 
  test "legacy turns remain explicit instead of receiving fabricated provenance" do
150 182 
    assert {:ok, conversation} = Conversations.ensure_conversation("legacy-browser")
151 183 
    assert {:ok, records} = Conversations.create_turn(conversation, "Legacy turn.")
test/openagents/voice/open_ai/call_client_test.exs modified +13

@@ -88,6 +88,19 @@ defmodule OpenAgents.Voice.OpenAI.CallClientTest do

88 88 
             )
89 89 
  end
90 90
91 
  test "call creation does not retry a failed POST without an idempotency contract" do
92 
    Req.Test.expect(__MODULE__, fn conn -> Plug.Conn.send_resp(conn, 503, "unavailable") end)
93
94 
    assert {:error, {:http_status, 503}} =
95 
             CallClient.create(
96 
               "v=0\r\no=no-retry-offer",
97 
               String.duplicate("d", 64),
98 
               enabled_config(),
99 
               api_key: "test-secret",
100 
               request_options: [plug: {Req.Test, __MODULE__}]
101 
             )
102 
  end
103
91 104 
  defp enabled_config do
92 105 
    Config.build!(
93 106 
      enabled: true,
test/openagents/voice_test.exs modified +2 -1

@@ -361,7 +361,8 @@ defmodule OpenAgents.VoiceTest do

361 361 
    {:ok, conversation} = Conversations.ensure_conversation("voice-recovery-browser")
362 362 
    {:ok, session} = Voice.admit_session(conversation, enabled_config())
363 363
364 
    assert :ok = Voice.recover_interrupted_sessions()
364 
    recovery = start_supervised!({OpenAgents.VoiceRecovery, []})
365 
    _state = :sys.get_state(recovery)
365 366
366 367 
    recovered = Repo.get!(Session, session.id)
367 368 
    assert recovered.status == "failed"
test/openagents/work_job_test.exs modified +209 -57

@@ -1,6 +1,6 @@

1 1 
defmodule OpenAgents.WorkJobTest do
2 2 
  use OpenAgents.DataCase
3 
  alias OpenAgents.{Conversations, Work}
3 
  alias OpenAgents.{Accounts, Conversations, Machines, Work, WorkRecovery}
4 4 
  alias OpenAgents.Conversations.Message
5 5 
  alias OpenAgents.Work.{Job, JobServer}
6 6

@@ -100,6 +100,7 @@ defmodule OpenAgents.WorkJobTest do

100 100 
    Process.exit(pid, :kill)
101 101 
    assert_receive {:DOWN, ^monitor, :process, ^pid, :killed}
102 102
103 
    tool_ref = Process.monitor(tool_pid)
103 104 
    send(tool_pid, :release_test_tool)
104 105
105 106 
    before_recovery = Work.get_job!(job.id)

@@ -107,7 +108,7 @@ defmodule OpenAgents.WorkJobTest do

107 108
108 109 
    # Recovery restarts the worker: it re-claims through the generation fence
109 110 
    # (adopt) and continues the work — the job is NOT finalized interrupted.
110 
    assert :ok = Work.recover_interrupted_jobs()
111 
    run_work_recovery()
111 112
112 113 
    # The restarted worker adopted the row through the fence (generation bumped)
113 114 
    # and drove the job to a REAL terminal state — never the buried

@@ -116,6 +117,8 @@ defmodule OpenAgents.WorkJobTest do

116 117 
    # which is still an honest terminal outcome produced by actual resumed
117 118 
    # work, not a burial.)
118 119 
    final = wait_for_terminal(job.id)
120 
    assert_receive {:DOWN, ^tool_ref, :process, ^tool_pid, _reason}
121 
    await_worker_exit(job.id)
119 122 
    assert final.generation > before_recovery.generation
120 123 
    refute final.status == "interrupted"
121 124 
    assert final.error_code != "runtime_restarted"

@@ -142,24 +145,37 @@ defmodule OpenAgents.WorkJobTest do

142 145 
    # No second adoption: the generation is unchanged.
143 146 
    assert undisturbed.generation == running.generation
144 147
148 
    tool_ref = Process.monitor(tool_pid)
145 149 
    send(tool_pid, :release_test_tool)
146 150 
    final = wait_for_terminal(job.id)
151 
    assert_receive {:DOWN, ^tool_ref, :process, ^tool_pid, _reason}
152 
    await_worker_exit(job.id)
147 153 
    refute final.status == "interrupted"
148 154 
  end
149 155
150 
  defp wait_for_terminal(job_id, attempts \\ 100) do
156 
  defp wait_for_terminal(job_id, timeout_ms \\ 10_000) do
151 157 
    job = Work.get_job!(job_id)
152 158
153 
    cond do
154 
      job.status in Job.terminal_statuses() ->
155 
        job
159 
    if job.status in Job.terminal_statuses() do
160 
      job
161 
    else
162 
      deadline = System.monotonic_time(:millisecond) + timeout_ms
163 
      await_terminal(job_id, job.status, deadline)
164 
    end
165 
  end
166
167 
  defp await_terminal(job_id, last_status, deadline) do
168 
    remaining = max(deadline - System.monotonic_time(:millisecond), 0)
156 169
157 
      attempts <= 0 ->
158 
        flunk("job never reached a terminal status (stuck #{job.status})")
170 
    receive do
171 
      {:work_job_updated, %Job{id: ^job_id, status: status} = job}
172 
      when status in ["completed", "failed", "interrupted", "budget_exhausted", "cancelled"] ->
173 
        job
159 174
160 
      true ->
161 
        Process.sleep(100)
162 
        wait_for_terminal(job_id, attempts - 1)
175 
      {:work_job_updated, %Job{id: ^job_id, status: status}} ->
176 
        await_terminal(job_id, status, deadline)
177 
    after
178 
      remaining -> flunk("job never reached a terminal status (stuck #{last_status})")
163 179 
    end
164 180 
  end
165 181

@@ -186,43 +202,29 @@ defmodule OpenAgents.WorkJobTest do

186 202 
  end
187 203
188 204 
  test "an interrupted delegation job reports honestly, not with deep-work text" do
189 
    {:ok, conversation} = Conversations.ensure_conversation("deleg-report-browser")
190 
    owner = Conversations.get_conversation_owner!(conversation)
191
192 
    {:ok, job} =
193 
      Work.create_job(%{
194 
        conversation_id: conversation.id,
195 
        owner_visitor_id: owner.id,
196 
        surface: "text",
197 
        goal: "Delegate to claude on devin-test: refactor the input bar",
198 
        kind: "delegation",
199 
        delegation: %{"agent_id" => "claude", "machine_name" => "devin-test"}
200 
      })
205 
    {_conversation, job} =
206 
      create_delegation_job(
207 
        "deleg-report-browser",
208 
        "Delegate to claude on devin-test: refactor the input bar"
209 
      )
201 210
202 211 
    {:ok, finished} = Work.finish_job(job.id, "interrupted", error_code: "runtime_restarted")
203 212
204 213 
    assert finished.status == "interrupted"
205 214
206 215 
    assert finished.report =~
207 
             "Delegation to claude on devin-test was interrupted by a server restart"
216 
             "Delegation to claude on devin-test-deleg-report-browser was interrupted by a server restart"
208 217
209 218 
    refute finished.report =~ "Deep work job"
210 219 
    refute finished.report =~ "no tool calls had completed"
211 220 
  end
212 221
213 222 
  test "cancel_job finishes a queued job as cancelled with an honest report" do
214 
    {:ok, conversation} = Conversations.ensure_conversation("deleg-cancel-browser")
215 
    owner = Conversations.get_conversation_owner!(conversation)
216
217 
    {:ok, job} =
218 
      Work.create_job(%{
219 
        conversation_id: conversation.id,
220 
        owner_visitor_id: owner.id,
221 
        surface: "text",
222 
        goal: "Delegate to claude on devin-test: refactor the input bar",
223 
        kind: "delegation",
224 
        delegation: %{"agent_id" => "claude", "machine_name" => "devin-test"}
225 
      })
223 
    {_conversation, job} =
224 
      create_delegation_job(
225 
        "deleg-cancel-browser",
226 
        "Delegate to claude on devin-test: refactor the input bar"
227 
      )
226 228
227 229 
    assert {:ok, finished} = Work.cancel_job(job.id)
228 230 
    assert finished.status == "cancelled"

@@ -231,18 +233,11 @@ defmodule OpenAgents.WorkJobTest do

231 233 
  end
232 234
233 235 
  test "checkpoint_delegation_session persists the session id only for the live generation" do
234 
    {:ok, conversation} = Conversations.ensure_conversation("deleg-ckpt-browser")
235 
    owner = Conversations.get_conversation_owner!(conversation)
236
237 
    {:ok, job} =
238 
      Work.create_job(%{
239 
        conversation_id: conversation.id,
240 
        owner_visitor_id: owner.id,
241 
        surface: "text",
242 
        goal: "Delegate to claude on devin-test: do work",
243 
        kind: "delegation",
244 
        delegation: %{"agent_id" => "claude", "machine_name" => "devin-test"}
245 
      })
236 
    {_conversation, job} =
237 
      create_delegation_job(
238 
        "deleg-ckpt-browser",
239 
        "Delegate to claude on devin-test: do work"
240 
      )
246 241
247 242 
    {:ok, running} = Work.claim_for_run(job.id)
248 243

@@ -265,6 +260,50 @@ defmodule OpenAgents.WorkJobTest do

265 260 
             Work.checkpoint_delegation_session(job.id, running.generation, "sess-late")
266 261 
  end
267 262
263 
  test "delegation execution identity is immutable except for its fenced session checkpoint" do
264 
    {_conversation, job} =
265 
      create_delegation_job(
266 
        "deleg-immutable-execution",
267 
        "Delegate to claude on devin-test: do work"
268 
      )
269
270 
    assert_raise Postgrex.Error, ~r/work job identity is immutable/, fn ->
271 
      job
272 
      |> Ecto.Changeset.change(
273 
        delegation: Map.put(job.delegation, "prompt", "replace the admitted prompt")
274 
      )
275 
      |> Repo.update!()
276 
    end
277
278 
    assert_raise Ecto.ConstraintError, ~r/work_jobs_delegation_identity/, fn ->
279 
      Repo.insert!(%Job{
280 
        conversation_id: job.conversation_id,
281 
        owner_visitor_id: job.owner_visitor_id,
282 
        machine_id: job.machine_id,
283 
        surface: "text",
284 
        goal: "A mismatched budget must fail",
285 
        kind: "delegation",
286 
        delegation: Map.put(job.delegation, "timeout_ms", 1),
287 
        authority_snapshot: job.authority_snapshot,
288 
        budget_snapshot: job.budget_snapshot
289 
      })
290 
    end
291
292 
    assert_raise Postgrex.Error, ~r/work job machine authority snapshot mismatch/, fn ->
293 
      Repo.insert!(%Job{
294 
        conversation_id: job.conversation_id,
295 
        owner_visitor_id: job.owner_visitor_id,
296 
        machine_id: job.machine_id,
297 
        surface: "text",
298 
        goal: "A false machine authority snapshot must fail",
299 
        kind: "delegation",
300 
        delegation: job.delegation,
301 
        authority_snapshot: Map.put(job.authority_snapshot, "roots", ["/tmp/foreign-root"]),
302 
        budget_snapshot: job.budget_snapshot
303 
      })
304 
    end
305 
  end
306
268 307 
  test "boot recovery records a degraded incident for each interrupted job" do
269 308 
    {:ok, user} =
270 309 
      OpenAgents.Accounts.upsert_github_user(%{

@@ -275,23 +314,26 @@ defmodule OpenAgents.WorkJobTest do

275 314
276 315 
    {:ok, conversation} = Conversations.ensure_conversation(user)
277 316 
    owner = Conversations.get_conversation_owner!(conversation)
317 
    machine = machine_for(user, "deleg-recover")
318 
    :ok = Work.subscribe(conversation.id)
278 319
279 320 
    {:ok, job} =
280 
      Work.create_job(%{
281 
        conversation_id: conversation.id,
282 
        owner_visitor_id: owner.id,
283 
        surface: "text",
284 
        goal: "Delegate to claude on devin-test: do work",
285 
        kind: "delegation",
286 
        delegation: %{"agent_id" => "claude", "machine_name" => "devin-test"}
287 
      })
321 
      Work.create_job(
322 
        delegation_attributes(
323 
          conversation,
324 
          owner,
325 
          machine,
326 
          "Delegate to claude on devin-test: do work"
327 
        )
328 
      )
288 329
289 
    :ok = Work.recover_interrupted_jobs()
330 
    run_work_recovery()
290 331
291 332 
    # Resume-first recovery still restarts this delegation's worker; with no
292 333 
    # machine connected it finishes honestly (machine_offline) — let it settle
293 334 
    # so the async worker never outlives the test's DB sandbox.
294 335 
    final = wait_for_terminal(job.id)
336 
    await_worker_exit(job.id)
295 337 
    refute final.status == "interrupted"
296 338
297 339 
    incidents = OpenAgents.Incidents.list_recent(owner.user_id)

@@ -302,9 +344,41 @@ defmodule OpenAgents.WorkJobTest do

302 344 
    assert incident.surface == "delegation"
303 345 
  end
304 346
347 
  test "the database rejects a delegated machine owned by another account" do
348 
    {_conversation, admitted} =
349 
      create_delegation_job(
350 
        "deleg-owner-boundary",
351 
        "Delegate to claude on devin-test: preserve the owner boundary"
352 
      )
353
354 
    {:ok, outsider} =
355 
      Accounts.upsert_github_user(%{
356 
        github_id: System.unique_integer([:positive]),
357 
        github_login: "deleg-owner-outsider",
358 
        github_avatar_url: "https://avatars.githubusercontent.com/u/2?v=4"
359 
      })
360
361 
    foreign_machine = machine_for(outsider, "deleg-owner-outsider")
362
363 
    assert_raise Postgrex.Error, ~r/work job machine owner mismatch/, fn ->
364 
      Repo.insert!(%Job{
365 
        conversation_id: admitted.conversation_id,
366 
        owner_visitor_id: admitted.owner_visitor_id,
367 
        machine_id: foreign_machine.id,
368 
        surface: "text",
369 
        goal: "Cross-account delegation must fail",
370 
        kind: "delegation",
371 
        delegation: Map.put(admitted.delegation, "machine_id", foreign_machine.id),
372 
        authority_snapshot: admitted.authority_snapshot,
373 
        budget_snapshot: admitted.budget_snapshot
374 
      })
375 
    end
376 
  end
377
305 378 
  defp create_job(browser_key, goal) do
306 379 
    assert {:ok, conversation} = Conversations.ensure_conversation(browser_key)
307 380 
    owner = Conversations.get_conversation_owner!(conversation)
381 
    :ok = Work.subscribe(conversation.id)
308 382
309 383 
    assert {:ok, job} =
310 384 
             Work.create_job(%{

@@ -317,10 +391,88 @@ defmodule OpenAgents.WorkJobTest do

317 391 
    {conversation, job}
318 392 
  end
319 393
394 
  defp create_delegation_job(key, goal) do
395 
    {:ok, user} =
396 
      Accounts.upsert_github_user(%{
397 
        github_id: System.unique_integer([:positive]),
398 
        github_login: key,
399 
        github_avatar_url: "https://avatars.githubusercontent.com/u/1?v=4"
400 
      })
401
402 
    {:ok, conversation} = Conversations.ensure_conversation(user)
403 
    owner = Conversations.get_conversation_owner!(conversation)
404 
    :ok = Work.subscribe(conversation.id)
405 
    machine = machine_for(user, key)
406 
    {:ok, job} = Work.create_job(delegation_attributes(conversation, owner, machine, goal))
407 
    {conversation, job}
408 
  end
409
410 
  defp machine_for(user, key) do
411 
    {:ok, %{code: code}} =
412 
      Machines.start_pairing(%{
413 
        "name" => "devin-test-#{key}",
414 
        "tier" => "curated",
415 
        "platform" => "linux-x64",
416 
        "agent_version" => "0.1.0",
417 
        "roots" => ["/tmp/openagents-work"]
418 
      })
419
420 
    {:ok, machine} = Machines.approve_pairing(user, code)
421 
    machine
422 
  end
423
424 
  defp delegation_attributes(conversation, owner, machine, goal) do
425 
    %{
426 
      conversation_id: conversation.id,
427 
      owner_visitor_id: owner.id,
428 
      machine_id: machine.id,
429 
      surface: "text",
430 
      goal: goal,
431 
      kind: "delegation",
432 
      delegation: %{
433 
        "agent_id" => "claude",
434 
        "machine_id" => machine.id,
435 
        "machine_name" => machine.name,
436 
        "prompt" => "do work",
437 
        "cwd" => "/tmp/openagents-work",
438 
        "timeout_ms" => 3_600_000
439 
      },
440 
      authority_snapshot: %{
441 
        "machine_tier" => machine.tier,
442 
        "roots" => machine.roots,
443 
        "cwd" => "/tmp/openagents-work",
444 
        "agent_id" => "claude",
445 
        "machine_name" => machine.name
446 
      },
447 
      budget_snapshot: %{
448 
        "wall_clock_ms" => 3_600_000,
449 
        "maximum_prompt_bytes" => 8_000,
450 
        "maximum_report_bytes" => 8_000
451 
      }
452 
    }
453 
  end
454
320 455 
  defp run_job_to_exit(%Job{} = job, timeout \\ 5_000) do
321 456 
    pid = start_supervised!({JobServer, job.id})
322 457 
    monitor = Process.monitor(pid)
323 458 
    assert_receive {:DOWN, ^monitor, :process, ^pid, :normal}, timeout
324 459 
    :ok
325 460 
  end
461
462 
  defp await_worker_exit(job_id) do
463 
    case Horde.Registry.lookup(OpenAgents.HordeRegistry, {:work_job, job_id}) do
464 
      [{pid, _value}] ->
465 
        ref = Process.monitor(pid)
466 
        assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
467
468 
      [] ->
469 
        :ok
470 
    end
471 
  end
472
473 
  defp run_work_recovery do
474 
    pid = start_supervised!({WorkRecovery, []})
475 
    _state = :sys.get_state(pid)
476 
    :ok
477 
  end
326 478 
end
test/openagents_web/controllers/computer_agent_jobs_controller_test.exs modified +8

@@ -67,6 +67,14 @@ defmodule OpenAgentsWeb.ComputerAgentJobsControllerTest do

67 67 
    assert_receive {:DOWN, ^job_ref, :process, _pid, :normal}, 1_000
68 68 
    job = Work.get_job!(job_id)
69 69 
    assert job.status == "completed"
70 
    assert job.machine_id == machine.id
71 
    assert job.authority_snapshot["machine_tier"] == machine.tier
72 
    assert job.authority_snapshot["roots"] == machine.roots
73 
    assert job.authority_snapshot["cwd"] == @root
74 
    assert job.authority_snapshot["agent_id"] == "codex"
75 
    assert job.authority_snapshot["machine_name"] == machine.name
76 
    assert job.budget_snapshot["wall_clock_ms"] == 3_600_000
77 
    assert job.budget_snapshot["maximum_report_bytes"] == 8_000
70 78 
    assert job.report =~ "connected"
71 79 
    assert job.report =~ "Model: gpt-5.6-sol · Reasoning: medium · Mode: agent-full-access"
72 80 
    assert is_binary(job.report_message_id)
test/support/memory/semantic_worker_test_provider.ex added +25

@@ -0,0 +1,25 @@

1 
defmodule OpenAgents.Memory.SemanticWorkerTestProvider do
2 
  @moduledoc false
3
4 
  @behaviour OpenAgents.Memory.EmbeddingProvider
5
6 
  @impl true
7 
  def embed(_text, %{dimensions: dimensions}) do
8 
    case Application.fetch_env!(:openagents, :semantic_worker_test_mode) do
9 
      :success ->
10 
        {:ok, List.duplicate(0.0, dimensions)}
11
12 
      :failure ->
13 
        {:error, :semantic_provider_offline}
14
15 
      {:block, observer} when is_pid(observer) ->
16 
        send(observer, {:semantic_provider_started, self()})
17
18 
        receive do
19 
          :release_semantic_provider -> {:ok, List.duplicate(0.0, dimensions)}
20 
        after
21 
          60_000 -> {:error, :semantic_provider_timeout}
22 
        end
23 
    end
24 
  end
25 
end

This page updates live while a promote is in flight · changelog