diff --git a/docs/mcp-reference.json b/docs/mcp-reference.json
index 17b0cc6a..5a9f11e7 100644
--- a/docs/mcp-reference.json
+++ b/docs/mcp-reference.json
@@ -141,6 +141,7 @@
"work_agreement_resolve",
"work_agreement_revise",
"work_agreement_submit",
+ "workflow_budget_resume",
"workflow_launch_approve",
"workflow_launch_reject",
"workflow_retry_status"
@@ -1656,7 +1657,7 @@
"behavior": {
"access": "Main endpoint capability: operator or caretaker routine. Specialists and temporary agents are refused. The handler has no narrower caller-identity check.",
"effects": "None.",
- "notes": "group optionally filters needs_you, watching, working, scheduled or quiet. Unknown group values return no matching groups rather than a validation error. Signal order is the fleet's current attention ranking.",
+ "notes": "group optionally filters needs_you, watching, working, scheduled or quiet. Unknown group values return no matching groups rather than a validation error. Signal order is the fleet's current attention ranking. A workflow_rail signal's resume_run args carry run and the exact pause id; a run with no known pause id offers only open_workflows. Its raised_at is null when the pause time is unknown.",
"result": "groups array, each with group and signals. Signals contain subject, kind, group, urgency, headline, detail, raised_at and resolving actions (label, op, args).",
"summary": "Read the fleet attention ranking grouped by need."
},
@@ -6277,6 +6278,101 @@
],
"name": "work_agreement_submit"
},
+ {
+ "behavior": {
+ "access": "Main endpoint, verified authenticated human operator only. The tool checks the verified identity before calling shared Operator.Actions, which enforces human authority again. Caretakers, routines, subagents and absent identity refuse without mutations; no default MCP caller identity.",
+ "effects": "Exact run and pause lookup, rail raise and resume admission in the shared operation. The tool call does not invoke a model; queued nodes may execute asynchronously.",
+ "notes": "Input run_id (1 to 1024 characters) and pause_id (1 to 160 characters) are both required; pause_id is the pause arg of the workflow_rail signal's resume_run action in list_attention. The latest pause is never substituted for a missing or different one. Repeating the exact pause returns the original receipt, even after the run pauses again, completes or fails; it never raises the rail twice and may only recover a still-pending initial dispatch. A pause that is no longer current refuses stale_pause; a run that is not parked refuses not_paused. Refusal codes are human_identity_required, pause_required, no_such_run, stale_pause, not_paused, outer_transaction_unsupported and admission_failed. Errors omit private paths, raw exceptions and secrets.",
+ "result": "custode.workflow_budget_resume.v1 receipt: schema_version, run_id, pause_id, decision resumed, previous_budget_usd and budget_usd (null when unbounded) and status running. The receipt describes the run at admission. Admission is not execution, completion or subprocess settlement.",
+ "summary": "Raise the rail of one exact workflow budget pause and admit its run to continue."
+ },
+ "category": "operator",
+ "definition": {
+ "annotations": {},
+ "description": "Raise the budget rail of one exact workflow budget pause and resume its run as the verified human. Success is admission; work may continue asynchronously and completion is not guaranteed. Repeating the same pause returns the original receipt and never raises the rail twice.",
+ "inputSchema": {
+ "additionalProperties": false,
+ "properties": {
+ "pause_id": {
+ "maxLength": 160,
+ "minLength": 1,
+ "type": "string"
+ },
+ "run_id": {
+ "maxLength": 1024,
+ "minLength": 1,
+ "type": "string"
+ }
+ },
+ "required": [
+ "pause_id",
+ "run_id"
+ ],
+ "type": "object"
+ },
+ "name": "workflow_budget_resume",
+ "outputSchema": {
+ "additionalProperties": false,
+ "properties": {
+ "budget_usd": {
+ "type": [
+ "number",
+ "null"
+ ]
+ },
+ "decision": {
+ "enum": [
+ "resumed"
+ ],
+ "type": "string"
+ },
+ "pause_id": {
+ "maxLength": 160,
+ "minLength": 1,
+ "type": "string"
+ },
+ "previous_budget_usd": {
+ "type": [
+ "number",
+ "null"
+ ]
+ },
+ "run_id": {
+ "maxLength": 1024,
+ "minLength": 1,
+ "type": "string"
+ },
+ "schema_version": {
+ "enum": [
+ "custode.workflow_budget_resume.v1"
+ ],
+ "type": "string"
+ },
+ "status": {
+ "enum": [
+ "running"
+ ],
+ "type": "string"
+ }
+ },
+ "required": [
+ "budget_usd",
+ "decision",
+ "pause_id",
+ "previous_budget_usd",
+ "run_id",
+ "schema_version",
+ "status"
+ ],
+ "type": "object"
+ },
+ "title": "workflow_budget_resume"
+ },
+ "endpoints": [
+ "/mcp"
+ ],
+ "name": "workflow_budget_resume"
+ },
{
"behavior": {
"access": "Main endpoint, verified authenticated human operator only. Shared Operator.Actions enforces human authority. Caretakers, routines, subagents and absent identity refuse without mutations; no default MCP caller identity.",
diff --git a/docs/mcp-reference.md b/docs/mcp-reference.md
index 534ec635..b774af43 100644
--- a/docs/mcp-reference.md
+++ b/docs/mcp-reference.md
@@ -89,7 +89,7 @@ Routine instructions, prompt files, the `prompt_agent` tool, and an agent's stru
| Endpoint | Server | Tools | Resources | Templates | Prompts |
| --- | --- | ---: | ---: | ---: | ---: |
-| `/mcp` | custode 0.3.0 | 106 | 4 | 13 | 0 |
+| `/mcp` | custode 0.3.0 | 107 | 4 | 13 | 0 |
| `/mcp/memory` | memory 0.3.0 | 7 | 0 | 0 | 0 |
**`/mcp`**: protocol versions 2026-07-28, 2025-11-25, 2025-06-18; capabilities `{"resources":{},"tools":{}}`.
@@ -208,6 +208,7 @@ Categories are descriptive policy metadata, not an authorization guarantee.
| [work_agreement_resolve](#tool-work_agreement_resolve) | operator |
| [work_agreement_revise](#tool-work_agreement_revise) | operator |
| [work_agreement_submit](#tool-work_agreement_submit) | self_write |
+| [workflow_budget_resume](#tool-workflow_budget_resume) | operator |
| [workflow_launch_approve](#tool-workflow_launch_approve) | operator |
| [workflow_launch_reject](#tool-workflow_launch_reject) | operator |
| [workflow_retry_status](#tool-workflow_retry_status) | read |
@@ -789,7 +790,7 @@ Read the fleet attention ranking grouped by need.
**Access:** Main endpoint capability: operator or caretaker routine. Specialists and temporary agents are refused. The handler has no narrower caller-identity check.
-**Behavior, defaults and errors:** group optionally filters needs_you, watching, working, scheduled or quiet. Unknown group values return no matching groups rather than a validation error. Signal order is the fleet's current attention ranking.
+**Behavior, defaults and errors:** group optionally filters needs_you, watching, working, scheduled or quiet. Unknown group values return no matching groups rather than a validation error. Signal order is the fleet's current attention ranking. A workflow_rail signal's resume_run args carry run and the exact pause id; a run with no known pause id offers only open_workflows. Its raised_at is null when the pause time is unknown.
### Tool: list_disowned
@@ -2410,6 +2411,85 @@ Retain a revision-bound result and its evidence limits.
**Behavior, defaults and errors:** agreement_id, agreement_revision, matching assignment_id, summary, nonempty criterion_evidence and verification_limits are required. Each evidence entry has criterion_id, references (possibly empty) and note; criterion IDs must name unique criteria in that retained revision. outputs may be empty for useful negative findings and defaults to empty. Late results for a historical revision remain visible but cannot complete current intent or inherit acceptance. All mutations require a stable request_id for one logical operation. An authorized exact retry returns the original receipt with duplicate=true before stale-revision checks; different semantic payloads with the same caller/request_id conflict. Intent revision and record sequence are distinct. Unknown fields and duplicate atom/string keys are refused. IDs are nonblank strings up to 160 characters; prose up to 2000; reference values up to 2048 and labels up to 200; arrays up to 20 entries; total mutation payload up to 65536 bytes. References are opaque attributed links, never fetched, verified or used as access grants. No operation starts, resumes, cancels, schedules or redispatches work.
+### Tool: workflow_budget_resume
+
+Raise the rail of one exact workflow budget pause and admit its run to continue.
+
+**Endpoints:** /mcp. **Category:** operator.
+
+| Argument | Type | Schema required | Description | Other schema constraints |
+| --- | --- | --- | --- | --- |
+| pause_id | string | yes | | {"maxLength":160,"minLength":1} |
+| run_id | string | yes | | {"maxLength":1024,"minLength":1} |
+
+**Result:** custode.workflow_budget_resume.v1 receipt: schema_version, run_id, pause_id, decision resumed, previous_budget_usd and budget_usd (null when unbounded) and status running. The receipt describes the run at admission. Admission is not execution, completion or subprocess settlement.
+
+**Side effects:** Exact run and pause lookup, rail raise and resume admission in the shared operation. The tool call does not invoke a model; queued nodes may execute asynchronously.
+
+**Access:** Main endpoint, verified authenticated human operator only. The tool checks the verified identity before calling shared Operator.Actions, which enforces human authority again. Caretakers, routines, subagents and absent identity refuse without mutations; no default MCP caller identity.
+
+**Behavior, defaults and errors:** Input run_id (1 to 1024 characters) and pause_id (1 to 160 characters) are both required; pause_id is the pause arg of the workflow_rail signal's resume_run action in list_attention. The latest pause is never substituted for a missing or different one. Repeating the exact pause returns the original receipt, even after the run pauses again, completes or fails; it never raises the rail twice and may only recover a still-pending initial dispatch. A pause that is no longer current refuses stale_pause; a run that is not parked refuses not_paused. Refusal codes are human_identity_required, pause_required, no_such_run, stale_pause, not_paused, outer_transaction_unsupported and admission_failed. Errors omit private paths, raw exceptions and secrets.
+
+**Output schema:**
+
+```json
+{
+ "additionalProperties": false,
+ "properties": {
+ "budget_usd": {
+ "type": [
+ "number",
+ "null"
+ ]
+ },
+ "decision": {
+ "enum": [
+ "resumed"
+ ],
+ "type": "string"
+ },
+ "pause_id": {
+ "maxLength": 160,
+ "minLength": 1,
+ "type": "string"
+ },
+ "previous_budget_usd": {
+ "type": [
+ "number",
+ "null"
+ ]
+ },
+ "run_id": {
+ "maxLength": 1024,
+ "minLength": 1,
+ "type": "string"
+ },
+ "schema_version": {
+ "enum": [
+ "custode.workflow_budget_resume.v1"
+ ],
+ "type": "string"
+ },
+ "status": {
+ "enum": [
+ "running"
+ ],
+ "type": "string"
+ }
+ },
+ "required": [
+ "budget_usd",
+ "decision",
+ "pause_id",
+ "previous_budget_usd",
+ "run_id",
+ "schema_version",
+ "status"
+ ],
+ "type": "object"
+}
+```
+
### Tool: workflow_launch_approve
Approve an existing workflow proposal and admit its retained run.
diff --git a/docs/mcp/behavior.json b/docs/mcp/behavior.json
index 75b424ae..98cab383 100644
--- a/docs/mcp/behavior.json
+++ b/docs/mcp/behavior.json
@@ -271,7 +271,7 @@
"list_attention": {
"access": "Main endpoint capability: operator or caretaker routine. Specialists and temporary agents are refused. The handler has no narrower caller-identity check.",
"effects": "None.",
- "notes": "group optionally filters needs_you, watching, working, scheduled or quiet. Unknown group values return no matching groups rather than a validation error. Signal order is the fleet's current attention ranking.",
+ "notes": "group optionally filters needs_you, watching, working, scheduled or quiet. Unknown group values return no matching groups rather than a validation error. Signal order is the fleet's current attention ranking. A workflow_rail signal's resume_run args carry run and the exact pause id; a run with no known pause id offers only open_workflows. Its raised_at is null when the pause time is unknown.",
"result": "groups array, each with group and signals. Signals contain subject, kind, group, urgency, headline, detail, raised_at and resolving actions (label, op, args).",
"summary": "Read the fleet attention ranking grouped by need."
},
@@ -848,6 +848,13 @@
"effects": "Exact proposal lookup and rejection event commit in one immediate SQLite transaction. Postcommit notification is best effort. No run or model call.",
"result": "Bounded structured proposal_id and recorded decision rejected. The retained reason is not echoed.",
"notes": "Input proposal_id is required, 1 to 160 characters; reason is optional, at most 2000 characters. An omitted or blank reason records rejected via mcp. Unknown or week-old pending proposals refuse. Authorization never uses bounded display history. Rejection retry retains the original decision and reason; an existing approval refuses. Precommit failure rolls back the decision and leaves the proposal pending. Outer transactions are unsupported. Errors omit private paths, raw exceptions and secrets."
+ },
+ "workflow_budget_resume": {
+ "summary": "Raise the rail of one exact workflow budget pause and admit its run to continue.",
+ "access": "Main endpoint, verified authenticated human operator only. The tool checks the verified identity before calling shared Operator.Actions, which enforces human authority again. Caretakers, routines, subagents and absent identity refuse without mutations; no default MCP caller identity.",
+ "effects": "Exact run and pause lookup, rail raise and resume admission in the shared operation. The tool call does not invoke a model; queued nodes may execute asynchronously.",
+ "result": "custode.workflow_budget_resume.v1 receipt: schema_version, run_id, pause_id, decision resumed, previous_budget_usd and budget_usd (null when unbounded) and status running. The receipt describes the run at admission. Admission is not execution, completion or subprocess settlement.",
+ "notes": "Input run_id (1 to 1024 characters) and pause_id (1 to 160 characters) are both required; pause_id is the pause arg of the workflow_rail signal's resume_run action in list_attention. The latest pause is never substituted for a missing or different one. Repeating the exact pause returns the original receipt, even after the run pauses again, completes or fails; it never raises the rail twice and may only recover a still-pending initial dispatch. A pause that is no longer current refuses stale_pause; a run that is not parked refuses not_paused. Refusal codes are human_identity_required, pause_required, no_such_run, stale_pause, not_paused, outer_transaction_unsupported and admission_failed. Errors omit private paths, raw exceptions and secrets."
}
}
}
diff --git a/lib/custode/attention.ex b/lib/custode/attention.ex
index a5dfcb17..49fc568c 100644
--- a/lib/custode/attention.ex
+++ b/lib/custode/attention.ex
@@ -269,9 +269,10 @@ defmodule Custode.Attention do
* `:launches` -- each as `%{id:, workflow:, repo:, summary:, why:,
proposed_at:}`.
- * `:paused_runs` -- each as `%{run_id:, workflow:, repo:, reason:,
- paused_at:}`, the reason being the run's own note of what it spent and
- which nodes it did not run.
+ * `:paused_runs` -- each as `%{run_id:, pause_id:, workflow:, repo:,
+ reason:, paused_at:}`, the reason being the run's own note of what it
+ spent and which nodes it did not run. `pause_id` is the exact pause a
+ resume must name; `paused_at` is nil when the pause time is unknown.
The subject is the workflow and the repository, which is how the operator
tells two of them apart; the ops carry the proposal or run id.
@@ -305,8 +306,21 @@ defmodule Custode.Attention do
# "Raise the rail and resume" rather than "Resume": resuming onto the same
# ceiling parks the run again on its next advance, so the one useful act is
- # the raise. `Custode.Workflow.Launch.raise_and_resume/1` owns by how much.
+ # the raise. `Custode.Workflow.BudgetResume` owns by how much.
+ #
+ # The op quotes the exact pause it was raised for (#846). A run with no
+ # usable pause id offers only the workflows page: resuming "whatever pause
+ # is current" is the guess the pause id exists to prevent.
defp workflow_rail(run) do
+ resume =
+ case run[:pause_id] do
+ pause when is_binary(pause) and pause != "" ->
+ [op("Raise the rail and resume", :resume_run, %{run: run.run_id, pause: pause})]
+
+ _unknown ->
+ []
+ end
+
%Signal{
subject: workflow_subject(run),
kind: :workflow_rail,
@@ -316,10 +330,7 @@ defmodule Custode.Attention do
detail: run[:reason],
item: {:run, run.run_id},
raised_at: run[:paused_at],
- resolving: [
- op("Raise the rail and resume", :resume_run, %{run: run.run_id}),
- op("Open workflows", :open_workflows, %{})
- ]
+ resolving: resume ++ [op("Open workflows", :open_workflows, %{})]
}
end
diff --git a/lib/custode/attention/fleet.ex b/lib/custode/attention/fleet.ex
index 10614244..732f61a8 100644
--- a/lib/custode/attention/fleet.ex
+++ b/lib/custode/attention/fleet.ex
@@ -135,33 +135,22 @@ defmodule Custode.Attention.Fleet do
}
end
- # The run row records that it is parked and why, not when: the pause is a
- # status change on a row with no timestamp for it. The `workflow_budget_paused`
- # feed entry is the record of when, and it is only read when a run is
- # actually parked, which is rare.
+ # The run row names its current pause and when it was taken (#846). The pause
+ # id is what a resume must quote, so a click on a row the operator saw cannot
+ # resume a later pause. A legacy pause has an id but no known time: nil, not
+ # a guess from bounded feed history.
defp paused_runs do
- case Run.list(status: "budget_paused") do
- [] ->
- []
-
- runs ->
- paused_at =
- "workflow_budget_paused"
- |> Custode.Feed.recent_by_event(limit: 50)
- |> Enum.reverse()
- |> Map.new(&{&1["run"], parse_at(&1["at"])})
-
- for run <- runs do
- %{
- run_id: run.run_id,
- workflow: run.workflow,
- repo: run.repo,
- # the pause note says what was spent AND which nodes did not run;
- # `error` is the same sentence without the second half
- reason: List.last(run.notes) || run.error,
- paused_at: Map.get(paused_at, run.run_id)
- }
- end
+ for run <- Run.list(status: "budget_paused") do
+ %{
+ run_id: run.run_id,
+ pause_id: run.budget_pause_id,
+ workflow: run.workflow,
+ repo: run.repo,
+ # the pause note says what was spent AND which nodes did not run;
+ # `error` is the same sentence without the second half
+ reason: List.last(run.notes) || run.error,
+ paused_at: run.budget_paused_at
+ }
end
end
diff --git a/lib/custode/mcp/server.ex b/lib/custode/mcp/server.ex
index 1d09854d..741dd300 100644
--- a/lib/custode/mcp/server.ex
+++ b/lib/custode/mcp/server.ex
@@ -62,6 +62,7 @@ defmodule Custode.MCP.Server do
Custode.MCP.WorkflowRetryTools.Read,
Custode.MCP.WorkflowDecisionTools.Approve,
Custode.MCP.WorkflowDecisionTools.Reject,
+ Custode.MCP.WorkflowBudgetTools.Resume,
Custode.MCP.ReadTools.ExecutingTurns,
Custode.MCP.ProjectProgressTools.Read,
Custode.MCP.WorkAgreementTools.Create,
diff --git a/lib/custode/mcp/tool_policy.ex b/lib/custode/mcp/tool_policy.ex
index f23428a6..c53f3262 100644
--- a/lib/custode/mcp/tool_policy.ex
+++ b/lib/custode/mcp/tool_policy.ex
@@ -118,6 +118,7 @@ defmodule Custode.MCP.ToolPolicy do
"workflow_retry_status" => :read,
"workflow_launch_approve" => :operator,
"workflow_launch_reject" => :operator,
+ "workflow_budget_resume" => :operator,
"executing_turns" => :read,
"operator_bootstrap" => :read,
"project_progress" => :read,
diff --git a/lib/custode/mcp/workflow_budget_tools.ex b/lib/custode/mcp/workflow_budget_tools.ex
new file mode 100644
index 00000000..43a36242
--- /dev/null
+++ b/lib/custode/mcp/workflow_budget_tools.ex
@@ -0,0 +1,101 @@
+defmodule Custode.MCP.WorkflowBudgetTools do
+ @moduledoc false
+
+ alias Custode.MCP.Tools
+ alias Custode.Operator.Actions
+
+ @schema_version "custode.workflow_budget_resume.v1"
+
+ @refusals [
+ :human_identity_required,
+ :pause_required,
+ :no_such_run,
+ :stale_pause,
+ :not_paused,
+ :outer_transaction_unsupported
+ ]
+
+ def input do
+ %{
+ "type" => "object",
+ "additionalProperties" => false,
+ "required" => ["run_id", "pause_id"],
+ "properties" => %{
+ "run_id" => %{"type" => "string", "minLength" => 1, "maxLength" => 1_024},
+ "pause_id" => %{"type" => "string", "minLength" => 1, "maxLength" => 160}
+ }
+ }
+ end
+
+ def output do
+ properties = %{
+ "schema_version" => %{"type" => "string", "enum" => [@schema_version]},
+ "run_id" => %{"type" => "string", "minLength" => 1, "maxLength" => 1_024},
+ "pause_id" => %{"type" => "string", "minLength" => 1, "maxLength" => 160},
+ "decision" => %{"type" => "string", "enum" => ["resumed"]},
+ "previous_budget_usd" => %{"type" => ["number", "null"]},
+ "budget_usd" => %{"type" => ["number", "null"]},
+ "status" => %{"type" => "string", "enum" => ["running"]}
+ }
+
+ %{
+ "type" => "object",
+ "additionalProperties" => false,
+ "required" => Map.keys(properties),
+ "properties" => properties
+ }
+ end
+
+ def execute(params, frame) do
+ case frame.assigns[:custode_identity] do
+ %{kind: :operator, id: id} = actor when is_binary(id) and id != "" ->
+ case Actions.resume_run(params.run_id, params.pause_id, actor: actor, via: :mcp) do
+ {:ok, receipt} -> success(frame, receipt)
+ {:error, reason} -> refusal(frame, reason)
+ end
+
+ _missing_or_nonhuman ->
+ refusal(frame, :human_identity_required)
+ end
+ end
+
+ # Only the receipt's known fields leave the server, so a core change cannot
+ # leak a private field through this tool.
+ defp success(frame, receipt) do
+ result = %{
+ "schema_version" => receipt.schema_version,
+ "run_id" => receipt.run_id,
+ "pause_id" => receipt.pause_id,
+ "decision" => receipt.decision,
+ "previous_budget_usd" => receipt.previous_budget_usd,
+ "budget_usd" => receipt.budget_usd,
+ "status" => receipt.status
+ }
+
+ {:reply, Snodo.Result.structured(result), frame}
+ end
+
+ defp refusal(frame, reason) do
+ code =
+ case reason do
+ code when code in @refusals -> Atom.to_string(code)
+ # Authority.human answers a human-readable string for a refused actor.
+ message when is_binary(message) -> "human_identity_required"
+ _private_failure -> "admission_failed"
+ end
+
+ Tools.fail(frame, "workflow budget resume refused: " <> code)
+ end
+end
+
+defmodule Custode.MCP.WorkflowBudgetTools.Resume do
+ @moduledoc "Raise the budget rail of one exact workflow budget pause and resume its run as the verified human. Success is admission; work may continue asynchronously and completion is not guaranteed. Repeating the same pause returns the original receipt and never raises the rail twice."
+ use Custode.MCP.Tool, name: "workflow_budget_resume", strict_arguments: true
+ alias Custode.MCP.WorkflowBudgetTools
+
+ input_schema(WorkflowBudgetTools.input())
+ output_schema(WorkflowBudgetTools.output())
+
+ @impl true
+ def execute(params, frame), do: WorkflowBudgetTools.execute(params, frame)
+end
diff --git a/lib/custode/operator/actions.ex b/lib/custode/operator/actions.ex
index 98a47a9c..fcc02759 100644
--- a/lib/custode/operator/actions.ex
+++ b/lib/custode/operator/actions.ex
@@ -26,6 +26,7 @@ defmodule Custode.Operator.Actions do
alias Custode.Operator.Authority
alias Custode.OperatorMessages
alias Custode.Suggestions
+ alias Custode.Workflow.BudgetResume
alias Custode.Workflow.Launch
@type result :: :ok | {:error, term()}
@@ -855,11 +856,20 @@ defmodule Custode.Operator.Actions do
end
@doc """
- Let a run parked on its budget rail go on, with the rail raised (#447). By
- how much is `Custode.Workflow.Launch.raise_and_resume/1`'s to say.
+ Admit a human budget increase for the exact displayed pause. Repeating the
+ reference returns its retained receipt without raising a later pause's rail.
"""
- @spec resume_run(String.t(), keyword()) :: {:ok, map()} | {:error, term()}
- def resume_run(run_id, _opts \\ []), do: Launch.raise_and_resume(run_id)
+ @spec resume_run(String.t(), String.t() | nil, keyword()) :: {:ok, map()} | {:error, term()}
+ def resume_run(run_id, pause_id, opts \\ []) do
+ with :ok <- Authority.human(actor(opts)) do
+ BudgetResume.resume(
+ run_id,
+ pause_id,
+ actor(opts),
+ Keyword.get(opts, :via, :liveview)
+ )
+ end
+ end
@doc """
Run a signal's `resolving` op. `params` carries what the op needs beyond its
@@ -903,7 +913,9 @@ defmodule Custode.Operator.Actions do
def run(:reject_launch, %{proposal: id}, params, opts),
do: reject_launch(id, params["reason"], opts)
- def run(:resume_run, %{run: id}, _params, opts), do: id |> resume_run(opts) |> outcome()
+ def run(:resume_run, %{run: id} = args, _params, opts),
+ do: id |> resume_run(Map.get(args, :pause), opts) |> outcome()
+
def run(op, _args, _params, _opts), do: {:error, {:unhandled_op, op}}
# `run/4` answers `:ok` or an error; the run a workflow op returns is for a
diff --git a/lib/custode/operator/workflow_budget.ex b/lib/custode/operator/workflow_budget.ex
new file mode 100644
index 00000000..ff284cc5
--- /dev/null
+++ b/lib/custode/operator/workflow_budget.ex
@@ -0,0 +1,20 @@
+defmodule Custode.Operator.WorkflowBudget do
+ @moduledoc "Shared operator wording for exact-pause workflow budget decisions."
+
+ def resumed(receipt) do
+ "Resume approved for this pause: budget #{budget(receipt.previous_budget_usd)} → " <>
+ "#{budget(receipt.budget_usd)}. Check the run below for its current status."
+ end
+
+ def refused(:pause_required), do: "No pause reference was supplied. Refresh and try again."
+
+ def refused(:stale_pause),
+ do: "This run has a different budget pause now. Review its current status before resuming."
+
+ def refused(:not_paused), do: "This run is no longer paused for its budget."
+ def refused(:no_such_run), do: "This workflow run no longer exists."
+ def refused(_reason), do: "The resume could not be approved. Refresh and try again."
+
+ defp budget(nil), do: "unlimited"
+ defp budget(amount), do: "$" <> :erlang.float_to_binary(amount * 1.0, decimals: 2)
+end
diff --git a/lib/custode/workflow/budget_resume.ex b/lib/custode/workflow/budget_resume.ex
new file mode 100644
index 00000000..29aa4c08
--- /dev/null
+++ b/lib/custode/workflow/budget_resume.ex
@@ -0,0 +1,106 @@
+defmodule Custode.Workflow.BudgetResume do
+ @moduledoc "Exact-pause, human-attributed budget admissions with retained retry receipts."
+
+ import Ecto.Query, only: [from: 2]
+ alias Custode.{Feed, Repo}
+ alias Custode.Workflow.{Run, Runner}
+
+ defmodule Row do
+ @moduledoc false
+ use Ecto.Schema
+
+ schema "workflow_budget_resumes" do
+ field(:run_id, :string)
+ field(:pause_id, :string)
+ field(:previous_budget_usd, :float)
+ field(:budget_usd, :float)
+ field(:actor_id, :string)
+ field(:via, :string)
+ field(:admitted_at, :utc_datetime_usec)
+ end
+ end
+
+ @doc "Admit once; receipt status describes admission, never current execution or settlement."
+ def resume(run_id, pause_id, actor, via) do
+ with :ok <- human(actor),
+ :ok <- reference(pause_id),
+ {:ok, {row, events}} <- admit(run_id, pause_id, actor.id, via) do
+ Enum.each(events, &Runner.publish_launch_event/1)
+ Runner.advance_budget_admitted(run_id, pause_id)
+ {:ok, receipt(row)}
+ end
+ end
+
+ defp human(%{kind: :operator, id: id}) when is_binary(id) and byte_size(id) > 0, do: :ok
+ defp human(_actor), do: {:error, :human_identity_required}
+
+ defp reference(id) when is_binary(id) and byte_size(id) > 0 and byte_size(id) <= 160, do: :ok
+ defp reference(_id), do: {:error, :pause_required}
+
+ defp admit(run_id, pause_id, actor_id, via) do
+ if Repo.in_transaction?() do
+ {:error, :outer_transaction_unsupported}
+ else
+ Repo.transaction(fn -> admit_locked(run_id, pause_id, actor_id, via) end, mode: :immediate)
+ end
+ rescue
+ _error -> {:error, :admission_failed}
+ end
+
+ defp admit_locked(run_id, pause_id, actor_id, via) do
+ run = Run.get(run_id) || Repo.rollback(:no_such_run)
+
+ case Repo.one(from(r in Row, where: r.run_id == ^run_id and r.pause_id == ^pause_id)) do
+ nil -> create_decision(run, pause_id, actor_id, via)
+ row -> {row, []}
+ end
+ end
+
+ defp create_decision(run, pause_id, actor_id, via) do
+ if run.status != "budget_paused", do: Repo.rollback(:not_paused)
+ if run.budget_pause_id != pause_id, do: Repo.rollback(:stale_pause)
+ budget = if is_nil(run.budget_usd), do: nil, else: run.budget_usd * 2
+
+ row =
+ Repo.insert!(%Row{
+ run_id: run.run_id,
+ pause_id: pause_id,
+ previous_budget_usd: run.budget_usd,
+ budget_usd: budget,
+ actor_id: actor_id,
+ via: to_string(via),
+ admitted_at: DateTime.utc_now()
+ })
+
+ Run.admit_budget_resume(run.run_id, pause_id, budget)
+
+ case Feed.record_in_transaction(%{
+ event: "workflow_resumed",
+ agent: nil,
+ run: run.run_id,
+ workflow: run.workflow,
+ repo: run.repo,
+ pause_id: pause_id,
+ previous_budget_usd: run.budget_usd,
+ budget_usd: budget,
+ actor: %{kind: "operator", id: actor_id},
+ via: to_string(via),
+ summary: "admitted budget resume for #{run.workflow} run #{run.run_id}"
+ }) do
+ {:ok, event} -> {row, [event]}
+ {:error, reason} -> Repo.rollback(reason)
+ end
+ end
+
+ defp receipt(row) do
+ %{
+ schema_version: "custode.workflow_budget_resume.v1",
+ run_id: row.run_id,
+ pause_id: row.pause_id,
+ decision: "resumed",
+ previous_budget_usd: row.previous_budget_usd,
+ budget_usd: row.budget_usd,
+ status: "running"
+ }
+ end
+end
diff --git a/lib/custode/workflow/launch.ex b/lib/custode/workflow/launch.ex
index 1ac20c97..18702c9b 100644
--- a/lib/custode/workflow/launch.ex
+++ b/lib/custode/workflow/launch.ex
@@ -329,6 +329,8 @@ defmodule Custode.Workflow.Launch do
end
@doc """
+ Low-level local control for explicit rail changes. Operator surfaces use
+ `Custode.Operator.Actions.resume_run/3` with an exact pause reference.
Let a budget-paused run go on, optionally on a raised rail. Records the
decision in the feed and advances the run -- resuming onto an unraised rail
parks it again immediately, which is the honest answer rather than a silent
@@ -361,23 +363,6 @@ defmodule Custode.Workflow.Launch do
end
end
- @doc """
- Let a parked run go on with its rail DOUBLED: the one act every surface
- offers for a budget pause (#447).
-
- Resuming onto the same ceiling parks the run again on its next advance, so
- the button raises the rail rather than pretending the limit is gone. The
- factor lives here because the workflows page and the inbox both offer the
- button, and two pages must not come to hold two ideas of what it does. A run
- with no rail is resumed as it is.
- """
- def raise_and_resume(run_id) do
- case Run.get(run_id) do
- %{budget_usd: budget} when is_number(budget) -> unpause(run_id, budget_usd: budget * 2)
- _no_rail_or_no_run -> unpause(run_id)
- end
- end
-
@doc """
Runs worth showing, newest first: everything still live, then the last few
finished ones. The checklist card reads this.
diff --git a/lib/custode/workflow/run.ex b/lib/custode/workflow/run.ex
index d040aa7b..c86ab0ed 100644
--- a/lib/custode/workflow/run.ex
+++ b/lib/custode/workflow/run.ex
@@ -48,6 +48,9 @@ defmodule Custode.Workflow.Run do
field(:notes, :string)
field(:error, :string)
field(:budget_usd, :float)
+ field(:budget_pause_id, :string)
+ field(:budget_paused_at, :utc_datetime_usec)
+ field(:budget_resume_pending, :string)
field(:work_item_id, :string)
field(:execution_generation, :string)
field(:definition_snapshot, :string)
@@ -126,7 +129,14 @@ defmodule Custode.Workflow.Run do
@doc "Mark a run finished. No stage cursor remains -- there is nothing next."
def complete(run_id) do
- update(run_id, status: "complete", stage: nil, finished_at: DateTime.utc_now())
+ update(run_id,
+ status: "complete",
+ stage: nil,
+ finished_at: DateTime.utc_now(),
+ budget_pause_id: nil,
+ budget_paused_at: nil,
+ budget_resume_pending: nil
+ )
end
@doc """
@@ -139,6 +149,9 @@ defmodule Custode.Workflow.Run do
status: "failed",
error: to_string(reason),
failure_identity: encode_optional(identity),
+ budget_pause_id: nil,
+ budget_paused_at: nil,
+ budget_resume_pending: nil,
finished_at: DateTime.utc_now()
)
end
@@ -152,8 +165,30 @@ defmodule Custode.Workflow.Run do
run says what it has NOT done rather than reading like a thin one.
"""
def budget_pause(run_id, reason, pending \\ []) do
- note(run_id, pause_note(reason, pending))
- update(run_id, status: "budget_paused", error: to_string(reason))
+ case get(run_id) do
+ nil ->
+ nil
+
+ run ->
+ note(run_id, pause_note(reason, pending))
+
+ same_pause? = run.status == "budget_paused" and is_binary(run.budget_pause_id)
+
+ pause_id =
+ if same_pause?,
+ do: run.budget_pause_id,
+ else: "wbp-" <> Base.encode16(:crypto.strong_rand_bytes(16), case: :lower)
+
+ paused_at = if same_pause?, do: run.budget_paused_at, else: DateTime.utc_now()
+
+ update(run_id,
+ status: "budget_paused",
+ error: to_string(reason),
+ budget_pause_id: pause_id,
+ budget_paused_at: paused_at,
+ budget_resume_pending: nil
+ )
+ end
end
defp pause_note(reason, []), do: "#{reason}; paused with no node outstanding"
@@ -172,11 +207,40 @@ defmodule Custode.Workflow.Run do
the honest outcome rather than a silent overrun.
"""
def unpause(run_id, budget_usd \\ :keep) do
- changes = [status: "running", error: nil]
+ changes = [
+ status: "running",
+ error: nil,
+ budget_pause_id: nil,
+ budget_paused_at: nil,
+ budget_resume_pending: nil
+ ]
+
changes = if budget_usd == :keep, do: changes, else: [{:budget_usd, budget_usd} | changes]
update(run_id, changes)
end
+ @doc "Persist an exact-pause admission inside its decision transaction."
+ def admit_budget_resume(run_id, pause_id, budget) do
+ unless Repo.in_transaction?(), do: raise(ArgumentError, "a transaction is required")
+
+ update(run_id,
+ status: "running",
+ error: nil,
+ budget_usd: budget,
+ budget_pause_id: nil,
+ budget_paused_at: nil,
+ budget_resume_pending: pause_id
+ )
+ end
+
+ @doc "Consume budget-resume dispatch evidence in the transaction that advances work."
+ def consume_budget_admission(%{budget_resume_pending: nil} = run), do: run
+
+ def consume_budget_admission(run) do
+ unless Repo.in_transaction?(), do: raise(ArgumentError, "a transaction is required")
+ update(run.run_id, budget_resume_pending: nil)
+ end
+
@doc """
The agent id a run's spend is booked under. The runner already stamps this
on every node job's meta, so `Custode.SpendLedger` records a run's cost
@@ -239,6 +303,9 @@ defmodule Custode.Workflow.Run do
notes: Jason.decode!(row.notes),
error: row.error,
budget_usd: row.budget_usd,
+ budget_pause_id: row.budget_pause_id,
+ budget_paused_at: row.budget_paused_at,
+ budget_resume_pending: row.budget_resume_pending,
work_item_id: row.work_item_id,
execution_generation: row.execution_generation,
definition_snapshot: decode_optional(row.definition_snapshot),
diff --git a/lib/custode/workflow/runner.ex b/lib/custode/workflow/runner.ex
index 407e2652..3d027d6c 100644
--- a/lib/custode/workflow/runner.ex
+++ b/lib/custode/workflow/runner.ex
@@ -200,6 +200,34 @@ defmodule Custode.Workflow.Runner do
end
end
+ @doc "Recover only the first dispatch of this exact budget-resume admission."
+ def advance_budget_admitted(run_id, pause_id) do
+ if Repo.in_transaction?() do
+ {:error, :outer_transaction_unsupported}
+ else
+ dispatch_budget_admitted(run_id, pause_id)
+ end
+ end
+
+ defp dispatch_budget_admitted(run_id, pause_id) do
+ case Repo.transaction(fn -> advance_budget_locked(run_id, pause_id) end, mode: :immediate) do
+ {:ok, result} -> result
+ _failed_advance -> admitted_state(run_id)
+ end
+ rescue
+ _error -> admitted_state(run_id)
+ catch
+ :exit, _reason -> admitted_state(run_id)
+ end
+
+ defp advance_budget_locked(run_id, pause_id) do
+ case Run.get(run_id) do
+ %{status: "running", budget_resume_pending: ^pause_id} -> advance_locked(run_id)
+ nil -> {:error, :no_such_run}
+ run -> {:ok, run}
+ end
+ end
+
defp dispatch_admitted(run_id) do
case Repo.transaction(fn -> advance_admitted_locked(run_id) end, mode: :immediate) do
{:ok, {:ok, run}} -> {:ok, run}
@@ -246,9 +274,17 @@ defmodule Custode.Workflow.Runner do
defp advance_locked(run_id) do
case Run.get(run_id) do
- nil -> {:error, :no_such_run}
- %{status: "running"} = run -> run |> Run.consume_launch_admission() |> advance_running()
- run -> {:ok, run}
+ nil ->
+ {:error, :no_such_run}
+
+ %{status: "running"} = run ->
+ run
+ |> Run.consume_launch_admission()
+ |> Run.consume_budget_admission()
+ |> advance_running()
+
+ run ->
+ {:ok, run}
end
end
diff --git a/lib/custode_web/console/item.ex b/lib/custode_web/console/item.ex
index 216183e0..008cc7ba 100644
--- a/lib/custode_web/console/item.ex
+++ b/lib/custode_web/console/item.ex
@@ -416,6 +416,20 @@ defmodule CustodeWeb.Console.Item do
"""
end
+ defp op(%{op: %{op: :resume_run}} = assigns) do
+ ~H"""
+ <.action_button
+ variant={:secondary}
+ phx-click="op"
+ phx-value-op="resume_run"
+ phx-value-run={@op.args.run}
+ phx-value-pause={@op.args.pause}
+ >
+ {@op.label}
+
+ """
+ end
+
defp op(assigns) do
~H"""
<.action_button
diff --git a/lib/custode_web/console_live.ex b/lib/custode_web/console_live.ex
index eefe19c4..75b2cd2a 100644
--- a/lib/custode_web/console_live.ex
+++ b/lib/custode_web/console_live.ex
@@ -39,6 +39,7 @@ defmodule CustodeWeb.ConsoleLive do
alias Custode.Operator.DirectoryBrowser
alias Custode.Operator.RoutineEdit
alias Custode.Operator.RoutineNew
+ alias Custode.Operator.WorkflowBudget
alias Custode.Signal
alias CustodeWeb.Console.Commands
alias CustodeWeb.Console.Rail
@@ -553,6 +554,19 @@ defmodule CustodeWeb.ConsoleLive do
|> after_action(socket, "rejected")
end
+ # A delayed click must name the pause that was rendered, even after a feed
+ # update replaces the selected signal. Missing references never use the
+ # current signal as a fallback; the shared action refuses them.
+ def handle_event("op", %{"op" => "resume_run"} = params, socket) do
+ case Actions.resume_run(params["run"], params["pause"], @opts) do
+ {:ok, receipt} ->
+ after_action(:ok, socket, WorkflowBudget.resumed(receipt))
+
+ {:error, reason} ->
+ {:noreply, socket |> assign(notice: WorkflowBudget.refused(reason)) |> refresh()}
+ end
+ end
+
# Any other op comes from the selected subject's own signal. The args are
# read back from that signal, never from the client, so a stale or forged
# click cannot name a different agent or action.
diff --git a/lib/custode_web/inbox_live.ex b/lib/custode_web/inbox_live.ex
index a3bc810e..c4a79734 100644
--- a/lib/custode_web/inbox_live.ex
+++ b/lib/custode_web/inbox_live.ex
@@ -21,6 +21,7 @@ defmodule CustodeWeb.InboxLive do
alias Custode.Operator.Actions
alias Custode.Operator.Inbox
+ alias Custode.Operator.WorkflowBudget
@impl Phoenix.LiveView
def mount(_params, _session, socket) do
@@ -92,13 +93,15 @@ defmodule CustodeWeb.InboxLive do
{:noreply, socket |> put_flash(:info, "launch rejected") |> refresh()}
end
- def handle_event("resume_run", %{"id" => id}, socket) do
- case Actions.resume_run(id, via: :liveview) do
- {:ok, _run} ->
- {:noreply, socket |> put_flash(:info, "run #{id} resumed on a raised rail") |> refresh()}
+ # The click names the pause the row showed (#846). Without one there is
+ # nothing to resume: the current pause is never picked in its place.
+ def handle_event("resume_run", %{"id" => id} = params, socket) do
+ case Actions.resume_run(id, params["pause"], via: :liveview) do
+ {:ok, receipt} ->
+ {:noreply, socket |> put_flash(:info, WorkflowBudget.resumed(receipt)) |> refresh()}
{:error, reason} ->
- {:noreply, socket |> put_flash(:error, "resume refused: #{inspect(reason)}") |> refresh()}
+ {:noreply, socket |> put_flash(:error, WorkflowBudget.refused(reason)) |> refresh()}
end
end
@@ -376,6 +379,7 @@ defmodule CustodeWeb.InboxLive do
variant={:primary}
phx-click="resume_run"
phx-value-id={@action.args[:run]}
+ phx-value-pause={@action.args[:pause]}
data-confirm="Double this run's rail and let it go on?"
>
{@action.label}
diff --git a/lib/custode_web/workflows_live.ex b/lib/custode_web/workflows_live.ex
index b0d4a89c..c13efbc9 100644
--- a/lib/custode_web/workflows_live.ex
+++ b/lib/custode_web/workflows_live.ex
@@ -26,6 +26,7 @@ defmodule CustodeWeb.WorkflowsLive do
import CustodeWeb.Components
alias Custode.Operator.Actions
+ alias Custode.Operator.WorkflowBudget
alias Custode.Workflow.Catalog
alias Custode.Workflow.Launch
alias Custode.Workflow.Results
@@ -61,15 +62,16 @@ defmodule CustodeWeb.WorkflowsLive do
end
# Letting a parked run go on is a rail RAISE, not a rail removal. By how
- # much is `Launch.raise_and_resume/1`'s to say, since the inbox offers the
- # same button (#447).
- def handle_event("resume_run", %{"id" => id}, socket) do
- case Actions.resume_run(id, via: :liveview) do
- {:ok, _run} ->
- {:noreply, socket |> put_flash(:info, "run #{id} resumed") |> refresh()}
+ # much is the shared budget-resume service's to say, since the inbox offers the
+ # same button (#447). The click names the pause the card showed (#846);
+ # without one nothing is resumed, and the current pause is never assumed.
+ def handle_event("resume_run", %{"id" => id} = params, socket) do
+ case Actions.resume_run(id, params["pause"], via: :liveview) do
+ {:ok, receipt} ->
+ {:noreply, socket |> put_flash(:info, WorkflowBudget.resumed(receipt)) |> refresh()}
{:error, reason} ->
- {:noreply, put_flash(socket, :error, "resume refused: #{inspect(reason)}")}
+ {:noreply, socket |> put_flash(:error, WorkflowBudget.refused(reason)) |> refresh()}
end
end
@@ -279,16 +281,26 @@ defmodule CustodeWeb.WorkflowsLive do
-
+
<.action_button
variant={:primary}
phx-click="resume_run"
phx-value-id={@entry.run.run_id}
+ phx-value-pause={pause_id(@entry.run)}
data-confirm="Double this run's rail and let it go on?"
>
Raise the rail and resume
+
+ This pause has no recorded id, so it cannot be resumed from here.
+
"""
@@ -326,6 +338,9 @@ defmodule CustodeWeb.WorkflowsLive do
defp stage_class(:not_run), do: "text-base-content/50"
defp stage_class(:unavailable), do: "text-error"
+ defp pause_id(%{budget_pause_id: pause}) when is_binary(pause) and pause != "", do: pause
+ defp pause_id(_run), do: nil
+
defp run_tone("running"), do: :info
defp run_tone("complete"), do: :success
defp run_tone("failed"), do: :error
diff --git a/priv/repo/migrations/20261012032630_add_workflow_budget_resume_decisions.exs b/priv/repo/migrations/20261012032630_add_workflow_budget_resume_decisions.exs
new file mode 100644
index 00000000..2dc1c37b
--- /dev/null
+++ b/priv/repo/migrations/20261012032630_add_workflow_budget_resume_decisions.exs
@@ -0,0 +1,34 @@
+defmodule Custode.Repo.Migrations.AddWorkflowBudgetResumeDecisions do
+ use Ecto.Migration
+
+ def change do
+ alter table(:workflow_runs) do
+ add(:budget_pause_id, :string)
+ add(:budget_paused_at, :utc_datetime_usec)
+ add(:budget_resume_pending, :string)
+ end
+
+ create table(:workflow_budget_resumes) do
+ add(
+ :run_id,
+ references(:workflow_runs, column: :run_id, type: :string, on_delete: :delete_all),
+ null: false
+ )
+
+ add(:pause_id, :string, null: false)
+ add(:previous_budget_usd, :float)
+ add(:budget_usd, :float)
+ add(:actor_id, :string, null: false)
+ add(:via, :string, null: false)
+ add(:admitted_at, :utc_datetime_usec, null: false)
+ end
+
+ create(unique_index(:workflow_budget_resumes, [:run_id, :pause_id]))
+
+ # Preserve unknown historical pause time. A reference grants no approval.
+ execute(
+ "UPDATE workflow_runs SET budget_pause_id = 'wbp-' || lower(hex(randomblob(16))) WHERE status = 'budget_paused'",
+ "UPDATE workflow_runs SET budget_pause_id = NULL"
+ )
+ end
+end
diff --git a/test/custode/attention_test.exs b/test/custode/attention_test.exs
index d9f4c803..bba5bac9 100644
--- a/test/custode/attention_test.exs
+++ b/test/custode/attention_test.exs
@@ -285,6 +285,7 @@ defmodule Custode.AttentionTest do
defp paused(run_id, fields \\ []) do
Enum.into(fields, %{
run_id: run_id,
+ pause_id: "pause-1",
workflow: "backlog-mine",
repo: "acme/widgets",
reason: "run budget rail hit: $1.20 of $1.00; paused before merge",
@@ -330,11 +331,29 @@ defmodule Custode.AttentionTest do
assert signal.raised_at == ~U[2026-07-26 01:30:00Z]
assert [
- %{label: "Raise the rail and resume", op: :resume_run, args: %{run: "run-9"}},
+ %{
+ label: "Raise the rail and resume",
+ op: :resume_run,
+ args: %{run: "run-9", pause: "pause-1"}
+ },
%{op: :open_workflows}
] = signal.resolving
end
+ test "a parked run with no known pause offers no resume, only the workflows page" do
+ for pause_id <- [nil, ""] do
+ assert [signal] =
+ Attention.workflows(%{
+ launches: [],
+ paused_runs: [paused("run-9", pause_id: pause_id, paused_at: nil)]
+ })
+
+ assert signal.kind == :workflow_rail
+ assert signal.raised_at == nil
+ assert [%{op: :open_workflows}] = signal.resolving
+ end
+ end
+
test "every proposal and every parked run is its own signal" do
facts = %{
launches: [launch("wfl-1"), launch("wfl-2", repo: "acme/gears")],
diff --git a/test/custode/mcp_workflow_budget_test.exs b/test/custode/mcp_workflow_budget_test.exs
new file mode 100644
index 00000000..e9c2274b
--- /dev/null
+++ b/test/custode/mcp_workflow_budget_test.exs
@@ -0,0 +1,335 @@
+defmodule Custode.MCPWorkflowBudgetTest do
+ use ExUnit.Case, async: false
+
+ import Custode.TestHelpers, only: [put_env!: 2, tmp_workspace!: 0, uid: 1]
+ import Ecto.Query, only: [from: 2]
+
+ alias Custode.{Feed, Repo, Workflow}
+ alias Custode.MCP.{CallContext, Capabilities, Identity, ToolPolicy, WorkflowBudgetTools}
+ alias Custode.Workflow.{Node, Run, Stage}
+
+ @tool "workflow_budget_resume"
+
+ setup do
+ name = uid("mcp-budget")
+ repo = uid("budget-repo")
+ worker = uid("budget-worker")
+ caretaker = uid("budget-caretaker")
+
+ workflow =
+ Workflow.new!(name, [
+ %Stage{
+ name: :fixture,
+ nodes: [%Node{name: :fixture, prompt: "fixture only", schema: %{}}]
+ }
+ ])
+
+ put_env!(:extra_workflows, %{name => workflow})
+
+ put_env!(
+ :routines,
+ for {id, role} <- [{worker, :backlog_worker}, {caretaker, :caretaker}] do
+ %{
+ id: id,
+ role: role,
+ cron: :manual,
+ workspace: tmp_workspace!(),
+ prompt: "fixture",
+ on_note: :ignore
+ }
+ end
+ )
+
+ on_exit(fn ->
+ run_ids = Repo.all(from(r in Run.Row, where: r.repo == ^repo, select: r.run_id))
+
+ Repo.delete_all(
+ from(j in Oban.Job,
+ where: fragment("json_extract(?, '$.workflow_run')", j.meta) in ^run_ids
+ )
+ )
+
+ Repo.delete_all(from(r in Run.Row, where: r.repo == ^repo))
+
+ Repo.delete_all(
+ from(f in Feed.Entry, where: fragment("json_extract(?, '$.repo')", f.entry) == ^repo)
+ )
+ end)
+
+ %{name: name, repo: repo, worker: worker, caretaker: caretaker}
+ end
+
+ defp parked!(ctx, budget_usd) do
+ run = Run.start(uid("budget-run"), ctx.name, ctx.repo, :fixture, %{}, budget_usd)
+ Run.budget_pause(run.run_id, "run budget rail hit: fixture", ["fixture"])
+ parked = Run.get(run.run_id)
+ assert parked.status == "budget_paused"
+ assert is_binary(parked.budget_pause_id)
+ parked
+ end
+
+ defp jobs(run_id) do
+ Repo.aggregate(
+ from(j in Oban.Job,
+ where: fragment("json_extract(?, '$.workflow_run')", j.meta) == ^run_id
+ ),
+ :count
+ )
+ end
+
+ test "the direct handler requires verified human identity with no MCP caller fallback", ctx do
+ run = parked!(ctx, 1.0)
+
+ for actor <- [
+ nil,
+ %{},
+ %{kind: :operator, id: ""},
+ %{kind: :routine, id: ctx.caretaker},
+ %{kind: :routine, id: ctx.worker},
+ %{kind: :sub_agent, id: uid("helper")}
+ ] do
+ frame = %CallContext{assigns: %{custode_identity: actor}}
+
+ assert {:reply, %Snodo.Result{kind: :error, value: message}, ^frame} =
+ WorkflowBudgetTools.Resume.execute(
+ %{run_id: run.run_id, pause_id: run.budget_pause_id},
+ frame
+ )
+
+ assert message == "workflow budget resume refused: human_identity_required"
+ end
+
+ unchanged = Run.get(run.run_id)
+ assert unchanged.status == "budget_paused"
+ assert unchanged.budget_usd == 1.0
+ assert unchanged.budget_pause_id == run.budget_pause_id
+ assert ToolPolicy.fetch(@tool) == {:ok, :operator}
+
+ for role <- [:caretaker, :backlog_worker] do
+ refute @tool in Capabilities.exposed_tool_names(role)
+ end
+ end
+
+ for protocol <- ["2025-11-25", "2026-07-28"] do
+ @protocol protocol
+ test "authenticated budget resume over HTTP #{@protocol}", ctx do
+ assert Application.get_env(:custode, :oban_queues) == []
+ human = Identity.mint(:operator, uid("budget-human"))
+ initialize(human, @protocol)
+
+ tool = Enum.find(discover(human, @protocol), &(&1["name"] == @tool))
+ assert tool
+ assert tool["inputSchema"] == WorkflowBudgetTools.Resume.input_schema()
+
+ assert MapSet.new(tool["outputSchema"]["required"]) ==
+ MapSet.new(WorkflowBudgetTools.Resume.output_schema()["required"])
+
+ assert tool["outputSchema"]["properties"] ==
+ WorkflowBudgetTools.Resume.output_schema()["properties"]
+
+ run = parked!(ctx, 1.0)
+ # Discover the resolving reference through the same HTTP surface an
+ # independent client uses, rather than relying on internal run fields.
+ attention = call(human, @protocol, "list_attention", %{"group" => "needs_you"})
+ assert %{"result" => %{"content" => [%{"text" => attention_json}]}} = attention
+ groups = Jason.decode!(attention_json)["groups"]
+
+ action =
+ groups
+ |> Enum.flat_map(& &1["signals"])
+ |> Enum.flat_map(& &1["resolving"])
+ |> Enum.find(&(&1["op"] == "resume_run" and &1["args"]["run"] == run.run_id))
+
+ assert action["args"]["pause"] == run.budget_pause_id
+ args = %{"run_id" => action["args"]["run"], "pause_id" => action["args"]["pause"]}
+
+ resumed = call(human, @protocol, @tool, args)
+ assert %{"result" => %{"isError" => false, "structuredContent" => receipt}} = resumed
+
+ assert receipt == %{
+ "schema_version" => "custode.workflow_budget_resume.v1",
+ "run_id" => run.run_id,
+ "pause_id" => run.budget_pause_id,
+ "decision" => "resumed",
+ "previous_budget_usd" => 1.0,
+ "budget_usd" => 2.0,
+ "status" => "running"
+ }
+
+ assert Run.get(run.run_id).budget_usd == 2.0
+ dispatched = jobs(run.run_id)
+
+ # An exact retry returns the admission receipt and raises nothing again,
+ # even once the run has parked on a new pause.
+ assert %{"result" => %{"structuredContent" => ^receipt}} =
+ call(human, @protocol, @tool, args)
+
+ Run.budget_pause(run.run_id, "run budget rail hit: again", ["fixture"])
+ repaused = Run.get(run.run_id)
+ refute repaused.budget_pause_id == run.budget_pause_id
+
+ assert %{"result" => %{"structuredContent" => ^receipt}} =
+ call(human, @protocol, @tool, args)
+
+ assert Run.get(run.run_id).status == "budget_paused"
+ assert Run.get(run.run_id).budget_usd == 2.0
+ assert jobs(run.run_id) == dispatched
+
+ # A pause the run never took is stale; the current one is not substituted.
+ assert_code(
+ call(human, @protocol, @tool, %{"run_id" => run.run_id, "pause_id" => uid("pause")}),
+ "stale_pause"
+ )
+
+ assert Run.get(run.run_id).budget_pause_id == repaused.budget_pause_id
+ assert Run.get(run.run_id).budget_usd == 2.0
+
+ assert_code(
+ call(human, @protocol, @tool, %{"run_id" => uid("unknown"), "pause_id" => uid("pause")}),
+ "no_such_run"
+ )
+
+ for arguments <- [
+ %{"run_id" => run.run_id},
+ %{"pause_id" => run.budget_pause_id},
+ %{"run_id" => "", "pause_id" => run.budget_pause_id},
+ %{"run_id" => run.run_id, "pause_id" => ""},
+ %{"run_id" => String.duplicate("x", 1_025), "pause_id" => "p"},
+ %{"run_id" => run.run_id, "pause_id" => String.duplicate("x", 161)},
+ %{"run_id" => run.run_id, "pause_id" => 1},
+ %{"run_id" => run.run_id, "pause_id" => "p", "budget_usd" => 99}
+ ] do
+ assert_invalid_arguments(human, @protocol, arguments)
+ end
+
+ assert_missing_arguments(human, @protocol)
+
+ pending = parked!(ctx, 1.0)
+ pending_args = %{"run_id" => pending.run_id, "pause_id" => pending.budget_pause_id}
+ before = Repo.aggregate(Feed.Entry, :count)
+
+ for id <- [ctx.worker, ctx.caretaker] do
+ token = Identity.mint(:routine, id)
+ initialize(token, @protocol)
+ refute @tool in Enum.map(discover(token, @protocol), & &1["name"])
+ assert_refused(call(token, @protocol, @tool, pending_args))
+ end
+
+ helper = Identity.mint(:sub_agent, uid("budget-helper"))
+
+ for token <- [helper, nil] do
+ response =
+ post(token, @protocol, "tools/call", %{"name" => @tool, "arguments" => pending_args})
+
+ assert response.status in [401, 403]
+ end
+
+ assert Repo.aggregate(Feed.Entry, :count) == before
+ untouched = Run.get(pending.run_id)
+ assert untouched.status == "budget_paused"
+ assert untouched.budget_usd == 1.0
+ assert untouched.budget_pause_id == pending.budget_pause_id
+ end
+ end
+
+ defp initialize(token, protocol) do
+ if protocol == "2025-11-25" do
+ response =
+ post(token, nil, "initialize", %{
+ "protocolVersion" => protocol,
+ "capabilities" => %{},
+ "clientInfo" => %{"name" => "workflow-fixture", "version" => "1"}
+ })
+
+ assert response.status == 200
+ assert response.body["result"]["protocolVersion"] == protocol
+ end
+ end
+
+ defp discover(token, protocol, params \\ %{}) do
+ response = post(token, protocol, "tools/list", params)
+ assert response.status == 200
+ page = response.body["result"]
+
+ case page["nextCursor"] do
+ nil -> page["tools"]
+ cursor -> page["tools"] ++ discover(token, protocol, %{"cursor" => cursor})
+ end
+ end
+
+ defp call(token, protocol, name, args) do
+ response = post(token, protocol, "tools/call", %{"name" => name, "arguments" => args})
+ assert response.status == 200
+ response.body
+ end
+
+ # A missing required argument is a tool result, as in the decision tools.
+ defp assert_missing_arguments(token, protocol) do
+ response = post(token, protocol, "tools/call", %{"name" => @tool, "arguments" => %{}})
+ assert response.status == 200
+
+ assert %{"result" => %{"isError" => true, "content" => [%{"text" => message}]}} =
+ response.body
+
+ assert message =~ "Missing required arguments"
+ end
+
+ defp assert_invalid_arguments(token, protocol, args) do
+ response = post(token, protocol, "tools/call", %{"name" => @tool, "arguments" => args})
+
+ case response.body do
+ %{"error" => %{"code" => -32_602}} ->
+ expected_status = if protocol == "2026-07-28", do: 400, else: 200
+ assert response.status == expected_status
+
+ %{"result" => %{"isError" => true, "content" => [%{"text" => message}]}} ->
+ assert response.status == 200
+ assert message =~ "Missing required arguments"
+ end
+ end
+
+ defp assert_code(response, code) do
+ assert %{"result" => %{"isError" => true, "content" => [%{"text" => message}]}} = response
+ assert message == "workflow budget resume refused: " <> code
+ end
+
+ defp assert_refused(%{"error" => _error}), do: :ok
+ defp assert_refused(%{"result" => %{"isError" => true}}), do: :ok
+
+ defp post(token, protocol, method, params) do
+ headers = [{"accept", "application/json, text/event-stream"}]
+ headers = if token, do: [{"authorization", "Bearer " <> token} | headers], else: headers
+ headers = if protocol, do: [{"mcp-protocol-version", protocol} | headers], else: headers
+
+ {headers, params} = modern_request(protocol, method, headers, params)
+
+ Req.post!(Custode.MCP.url(),
+ json: %{
+ "jsonrpc" => "2.0",
+ "id" => System.unique_integer([:positive]),
+ "method" => method,
+ "params" => params
+ },
+ headers: headers,
+ retry: false,
+ receive_timeout: 10_000
+ )
+ end
+
+ defp modern_request("2026-07-28" = protocol, method, headers, params) do
+ headers = [{"mcp-method", method} | headers]
+
+ headers =
+ if method == "tools/call", do: [{"mcp-name", params["name"]} | headers], else: headers
+
+ params =
+ Map.put(params, "_meta", %{
+ "io.modelcontextprotocol/protocolVersion" => protocol,
+ "io.modelcontextprotocol/clientCapabilities" => %{}
+ })
+
+ {headers, params}
+ end
+
+ defp modern_request(_protocol, _method, headers, params), do: {headers, params}
+end
diff --git a/test/custode/workflow_budget_migration_test.exs b/test/custode/workflow_budget_migration_test.exs
new file mode 100644
index 00000000..baad51e0
--- /dev/null
+++ b/test/custode/workflow_budget_migration_test.exs
@@ -0,0 +1,59 @@
+defmodule Custode.WorkflowBudgetMigrationTest do
+ use ExUnit.Case, async: false
+ import Custode.TestHelpers, only: [tmp_workspace!: 0]
+ alias Ecto.Adapters.SQL
+
+ defmodule MigrationRepo do
+ use Ecto.Repo, otp_app: :custode, adapter: Ecto.Adapters.SQLite3
+ end
+
+ test "migration gives existing pauses stable references without approvals or invented dates" do
+ directory = tmp_workspace!()
+
+ start_supervised!(
+ {MigrationRepo, database: Path.join(directory, "migration.db"), pool_size: 1}
+ )
+
+ query = &SQL.query!(MigrationRepo, &1, [])
+
+ query.(
+ "CREATE TABLE workflow_runs (id INTEGER PRIMARY KEY, run_id TEXT NOT NULL UNIQUE, status TEXT NOT NULL)"
+ )
+
+ query.(
+ "INSERT INTO workflow_runs (run_id, status) VALUES ('parked', 'budget_paused'), ('active', 'running'), ('done', 'complete')"
+ )
+
+ migration = Custode.Repo.Migrations.AddWorkflowBudgetResumeDecisions
+
+ unless Code.ensure_loaded?(migration) do
+ Code.require_file(
+ "priv/repo/migrations/20261012032630_add_workflow_budget_resume_decisions.exs"
+ )
+ end
+
+ version = 20_261_012_032_630
+ assert :ok = Ecto.Migrator.up(MigrationRepo, version, migration, log: false)
+
+ assert [["parked", pause, nil, nil]] =
+ query.(
+ "SELECT run_id, budget_pause_id, budget_paused_at, budget_resume_pending FROM workflow_runs WHERE status = 'budget_paused'"
+ ).rows
+
+ assert pause =~ ~r/^wbp-[0-9a-f]{32}$/
+ assert [[0]] = query.("SELECT count(*) FROM workflow_budget_resumes").rows
+
+ assert [[2]] =
+ query.(
+ "SELECT count(*) FROM workflow_runs WHERE status != 'budget_paused' AND budget_pause_id IS NULL"
+ ).rows
+
+ assert :already_up = Ecto.Migrator.up(MigrationRepo, version, migration, log: false)
+
+ assert [[^pause]] =
+ query.("SELECT budget_pause_id FROM workflow_runs WHERE run_id = 'parked'").rows
+
+ assert :ok = Ecto.Migrator.down(MigrationRepo, version, migration, log: false)
+ assert [[3]] = query.("SELECT count(*) FROM workflow_runs").rows
+ end
+end
diff --git a/test/custode/workflow_budget_resume_test.exs b/test/custode/workflow_budget_resume_test.exs
new file mode 100644
index 00000000..7918d1c0
--- /dev/null
+++ b/test/custode/workflow_budget_resume_test.exs
@@ -0,0 +1,537 @@
+defmodule Custode.WorkflowBudgetResumeTest do
+ use ExUnit.Case, async: false
+
+ import Custode.TestHelpers, only: [put_env!: 2, tmp_workspace!: 0, uid: 1]
+ import Ecto.Query, only: [from: 2]
+
+ alias Custode.{Feed, Repo, SpendLedger, Workflow}
+ alias Custode.Operator.Actions
+ alias Custode.Workflow.{BudgetResume, Node, Results, Run, Runner, Stage}
+
+ setup do
+ name = uid("budget-resume-workflow")
+ repo = uid("budget-resume-repo")
+ workspace = tmp_workspace!()
+
+ workflow =
+ Workflow.new!(name, [
+ %Stage{
+ name: :fixture,
+ nodes: [
+ %Node{
+ name: :fixture,
+ prompt: "fixture <%= Jason.encode!(@context) %>",
+ schema: %{}
+ }
+ ]
+ }
+ ])
+
+ put_env!(:extra_workflows, %{name => workflow})
+
+ on_exit(fn ->
+ run_ids = Repo.all(from(r in Run.Row, where: r.repo == ^repo, select: r.run_id))
+ spend_ids = Enum.map(run_ids, &Run.spend_agent_id/1)
+
+ Repo.delete_all(
+ from(j in Oban.Job,
+ where: fragment("json_extract(?, '$.workflow_run')", j.meta) in ^run_ids
+ )
+ )
+
+ Repo.delete_all(from(r in Results.Result, where: r.workflow_run in ^run_ids))
+ Repo.delete_all(from(r in BudgetResume.Row, where: r.run_id in ^run_ids))
+ Repo.delete_all(from(r in Run.Row, where: r.repo == ^repo))
+ Repo.delete_all(from(s in SpendLedger.Entry, where: s.agent_id in ^spend_ids))
+
+ Repo.delete_all(
+ from(f in Feed.Entry, where: fragment("json_extract(?, '$.run')", f.entry) in ^run_ids)
+ )
+ end)
+
+ assert Application.get_env(:custode, :oban_queues) == []
+ assert Application.get_env(:custode, :scheduler_autostart) == false
+ %{name: name, repo: repo, workspace: workspace, workflow: workflow}
+ end
+
+ test "one pause retains its reference until a genuine new pause", ctx do
+ paused = paused_run(ctx)
+ assert is_binary(paused.budget_pause_id)
+ assert %DateTime{} = paused.budget_paused_at
+ repeated = Run.budget_pause(paused.run_id, "same pause", ["fixture"])
+ assert repeated.budget_pause_id == paused.budget_pause_id
+ assert repeated.budget_paused_at == paused.budget_paused_at
+ assert repeated.budget_resume_pending == nil
+
+ unpaused = Run.unpause(paused.run_id)
+ assert_clear_admission(unpaused)
+ next = Run.budget_pause(paused.run_id, "new pause", ["fixture"])
+ assert next.budget_pause_id != paused.budget_pause_id
+ assert %DateTime{} = next.budget_paused_at
+ end
+
+ test "a legacy pause retains its unknown timestamp on repeated observation", ctx do
+ paused = paused_run(ctx)
+
+ Repo.update_all(from(r in Run.Row, where: r.run_id == ^paused.run_id),
+ set: [budget_paused_at: nil]
+ )
+
+ repeated = Run.budget_pause(paused.run_id, "same legacy pause", ["fixture"])
+ assert repeated.budget_pause_id == paused.budget_pause_id
+ assert repeated.budget_paused_at == nil
+ end
+
+ test "concurrent decisions retain one rail increase, receipt, event and job", ctx do
+ paused = paused_run(ctx)
+ actor = %{kind: :operator, id: uid("human")}
+
+ results =
+ concurrently(for _ <- 1..4, do: fn -> resume(paused, actor: actor, via: :mcp) end)
+
+ assert [{:ok, receipt} | _] = results
+ assert Enum.all?(results, &(&1 == {:ok, receipt}))
+ assert receipt == expected_receipt(paused, 2.0)
+ assert [%{actor_id: actor_id, via: "mcp", admitted_at: %DateTime{}}] = admissions(paused)
+ assert actor_id == actor.id
+ assert length(events(paused)) == 1
+ assert [job] = jobs(paused)
+ assert job.attempt == 0
+ assert job.state in ["available", "scheduled"]
+ retained = Run.get(paused.run_id)
+ assert retained.status == "running"
+ assert retained.budget_usd == 2.0
+ assert retained.context == paused.context
+ assert_clear_admission(retained)
+
+ assert {:ok, ^receipt} = resume(paused, actor: %{kind: :operator, id: uid("other-human")})
+ assert [%{actor_id: ^actor_id, via: "mcp"}] = admissions(paused)
+ assert length(events(paused)) == 1
+ end
+
+ test "an immediate second pause cannot be consumed by a lost-response retry", ctx do
+ paused = paused_run(ctx, spend: 5.0)
+ assert {:ok, first} = resume(paused)
+ assert first == expected_receipt(paused, 2.0)
+ second_pause = Run.get(paused.run_id)
+ assert second_pause.status == "budget_paused"
+ assert second_pause.budget_usd == 2.0
+ assert second_pause.budget_pause_id != paused.budget_pause_id
+ assert second_pause.budget_resume_pending == nil
+ assert jobs(paused) == []
+
+ assert {:ok, ^first} = resume(paused)
+ assert Run.get(paused.run_id) == second_pause
+ assert length(admissions(paused)) == 1
+ assert length(events(paused)) == 1
+
+ assert {:ok, second} = resume(second_pause)
+ assert second == expected_receipt(second_pause, 4.0)
+ third_pause = Run.get(paused.run_id)
+ assert third_pause.status == "budget_paused"
+ assert third_pause.budget_pause_id != second_pause.budget_pause_id
+ assert third_pause.budget_usd == 4.0
+ assert {:ok, ^first} = resume(paused)
+ assert {:ok, ^second} = resume(second_pause)
+ assert Run.get(paused.run_id) == third_pause
+
+ assert {:ok, third} = resume(third_pause)
+ assert third == expected_receipt(third_pause, 8.0)
+ assert Run.get(paused.run_id).status == "running"
+ assert length(jobs(paused)) == 1
+ assert length(admissions(paused)) == 3
+ assert length(events(paused)) == 3
+ end
+
+ test "an explicitly paused unbounded run remains unbounded", ctx do
+ paused = paused_run(ctx, budget: nil)
+ assert {:ok, receipt} = resume(paused)
+ assert receipt == expected_receipt(paused, nil)
+ assert Run.get(paused.run_id).budget_usd == nil
+ assert length(jobs(paused)) == 1
+ end
+
+ test "shared and internal operations refuse missing or nonhuman identity without effects",
+ ctx do
+ paused = paused_run(ctx)
+ before = snapshot(paused)
+
+ for actor <- [
+ nil,
+ %{},
+ %{kind: :routine, id: "caretaker"},
+ %{kind: :sub_agent, id: "helper"},
+ %{kind: :operator},
+ %{kind: :operator, id: ""}
+ ] do
+ assert {:error, _} = resume(paused, actor: actor)
+
+ assert {:error, :human_identity_required} =
+ BudgetResume.resume(paused.run_id, paused.budget_pause_id, actor, :mcp)
+
+ assert snapshot(paused) == before
+ end
+ end
+
+ test "missing, stale, unknown and nonpaused references cannot mutate work", ctx do
+ paused = paused_run(ctx)
+ before = snapshot(paused)
+
+ for invalid <- [nil, "", String.duplicate("x", 161)] do
+ assert {:error, :pause_required} = Actions.resume_run(paused.run_id, invalid)
+ end
+
+ assert {:error, :stale_pause} = Actions.resume_run(paused.run_id, uid("old-pause"))
+ assert {:error, :no_such_run} = Actions.resume_run(uid("missing-run"), paused.budget_pause_id)
+ assert snapshot(paused) == before
+
+ running = Run.unpause(paused.run_id)
+ assert {:error, :not_paused} = resume(paused)
+ assert Run.get(paused.run_id) == running
+ assert admissions(paused) == []
+ assert jobs(paused) == []
+ end
+
+ for terminal <- [:complete, :failed] do
+ @terminal terminal
+
+ test "#{terminal} clears pause authority and still returns a previously admitted receipt",
+ ctx do
+ paused = paused_run(ctx)
+ assert {:ok, receipt} = resume(paused)
+ terminal = finish(paused, @terminal)
+ assert_clear_admission(terminal)
+ retained_jobs = jobs(paused)
+ assert {:ok, ^receipt} = resume(paused)
+ assert Run.get(paused.run_id) == terminal
+ assert jobs(paused) == retained_jobs
+ assert length(admissions(paused)) == 1
+ assert length(events(paused)) == 1
+
+ never_admitted = paused_run(ctx)
+ finished = finish(never_admitted, @terminal)
+ assert_clear_admission(finished)
+ assert {:error, :not_paused} = resume(never_admitted)
+ assert Run.get(never_admitted.run_id) == finished
+ assert admissions(never_admitted) == []
+ assert jobs(never_admitted) == []
+ end
+ end
+
+ for job_state <- [:cancelled, :discarded, :pruned] do
+ @job_state job_state
+
+ test "retry never recreates #{@job_state} admitted work", ctx do
+ paused = paused_run(ctx)
+ assert {:ok, receipt} = resume(paused)
+ assert [job] = jobs(paused)
+
+ case @job_state do
+ :pruned ->
+ Repo.delete!(job)
+
+ state ->
+ Repo.update_all(from(j in Oban.Job, where: j.id == ^job.id),
+ set: [state: to_string(state)]
+ )
+ end
+
+ # The durable decision, not retained jobs or the optional feed, is authority.
+ Repo.delete_all(
+ from(f in Feed.Entry,
+ where: f.event == "workflow_resumed",
+ where: fragment("json_extract(?, '$.run')", f.entry) == ^paused.run_id
+ )
+ )
+
+ before = snapshot(paused)
+ assert {:ok, ^receipt} = resume(paused)
+ assert snapshot(paused) == before
+ assert Run.get(paused.run_id).budget_resume_pending == nil
+ end
+ end
+
+ test "budget admission never changes rendered context or planned argument hashes", ctx do
+ paused = paused_run(ctx)
+ assert [before] = Runner.plan(paused, ctx.workflow)
+ assert {:ok, _receipt} = resume(paused)
+ resumed = Run.get(paused.run_id)
+ assert resumed.context == paused.context
+ assert [after_resume] = Runner.plan(resumed, ctx.workflow)
+ assert after_resume.prompt == before.prompt
+ assert after_resume.args_hash == before.args_hash
+ assert [job] = jobs(paused)
+ assert job.args["prompt"] == before.prompt
+ assert job.meta["args_hash"] == before.args_hash
+
+ assert {:ok, _} = Runner.advance(paused.run_id)
+ assert {:ok, _} = resume(paused)
+ assert [same_job] = jobs(paused)
+ assert same_job.id == job.id
+ assert same_job.meta["args_hash"] == before.args_hash
+ end
+
+ test "shared resolving actions require the displayed pause and preserve human authority", ctx do
+ paused = paused_run(ctx)
+ args = %{run: paused.run_id, pause: paused.budget_pause_id}
+ assert {:error, :pause_required} = Actions.run(:resume_run, %{run: paused.run_id})
+
+ assert {:error, _} =
+ Actions.run(:resume_run, args, %{}, actor: %{kind: :routine, id: "caretaker"})
+
+ assert admissions(paused) == []
+ assert :ok = Actions.run(:resume_run, args, %{}, via: :liveview)
+ assert :ok = Actions.run(:resume_run, args, %{}, via: :liveview)
+ assert length(admissions(paused)) == 1
+ assert length(jobs(paused)) == 1
+ end
+
+ test "outer transactions refuse before admission or publication", ctx do
+ paused = paused_run(ctx)
+ before = snapshot(paused)
+
+ assert {:ok, :ok} =
+ Repo.transaction(fn ->
+ assert {:error, :outer_transaction_unsupported} = resume(paused)
+ :ok
+ end)
+
+ assert snapshot(paused) == before
+ end
+
+ test "a failed decision event rolls back the receipt, rail and marker without publication",
+ ctx do
+ mirror = Path.join(ctx.workspace, "feed.jsonl")
+ File.write!(mirror, "")
+ put_env!(:feed_path, mirror)
+ paused = paused_run(ctx)
+ before = snapshot(paused)
+ before_mirror = File.read!(mirror)
+ :ok = Custode.PubSubBridge.subscribe()
+
+ trigger =
+ abort_trigger!(
+ "feed_entries",
+ "NEW.event = 'workflow_resumed' AND json_extract(NEW.entry, '$.run') = '#{paused.run_id}'"
+ )
+
+ assert {:error, :admission_failed} = resume(paused)
+ assert snapshot(paused) == before
+ assert File.read!(mirror) == before_mirror
+ run_id = paused.run_id
+ refute_receive {:feed_entry, %{"event" => "workflow_resumed", "run" => ^run_id}}
+ drop_trigger!(trigger)
+ assert {:ok, _receipt} = resume(paused)
+ assert length(admissions(paused)) == 1
+ assert length(jobs(paused)) == 1
+ end
+
+ for recovery <- [:retry, :runner_resume] do
+ @recovery recovery
+
+ test "#{recovery} recovers only a committed admission whose dispatch transaction failed",
+ ctx do
+ paused = paused_run(ctx)
+
+ trigger =
+ abort_trigger!(
+ "oban_jobs",
+ "json_extract(NEW.meta, '$.workflow_run') = '#{paused.run_id}'"
+ )
+
+ assert {:ok, receipt} = resume(paused)
+ admitted = Run.get(paused.run_id)
+ assert admitted.status == "running"
+ assert admitted.budget_usd == 2.0
+ assert admitted.budget_resume_pending == paused.budget_pause_id
+ assert admitted.budget_pause_id == nil
+ assert jobs(paused) == []
+ assert length(admissions(paused)) == 1
+ assert length(events(paused)) == 1
+ drop_trigger!(trigger)
+
+ case @recovery do
+ :retry -> assert {:ok, ^receipt} = resume(paused)
+ :runner_resume -> assert {:ok, _resumed} = Runner.resume(paused.run_id)
+ end
+
+ assert [job] = jobs(paused)
+ assert_clear_admission(Run.get(paused.run_id))
+ Repo.delete!(job)
+ assert {:ok, ^receipt} = resume(paused)
+ assert jobs(paused) == []
+ assert length(admissions(paused)) == 1
+ assert length(events(paused)) == 1
+ end
+ end
+
+ test "a concurrent terminal transition cannot be undone by admission or its retry", ctx do
+ paused = paused_run(ctx)
+
+ [decision, terminal] =
+ concurrently([
+ fn -> resume(paused) end,
+ fn ->
+ Repo.transaction(fn -> Run.fail(paused.run_id, "fixture failure") end, mode: :immediate)
+ end
+ ])
+
+ assert {:ok, %{status: "failed"}} = terminal
+ failed = Run.get(paused.run_id)
+ assert failed.status == "failed"
+ assert_clear_admission(failed)
+ retained_jobs = jobs(paused)
+ assert length(retained_jobs) <= 1
+
+ case decision do
+ {:ok, receipt} ->
+ assert {:ok, ^receipt} = resume(paused)
+ assert failed.budget_usd == 2.0
+ assert length(admissions(paused)) == 1
+
+ {:error, :not_paused} ->
+ assert {:error, :not_paused} = resume(paused)
+ assert failed.budget_usd == 1.0
+ assert admissions(paused) == []
+ end
+
+ assert Run.get(paused.run_id) == failed
+ assert jobs(paused) == retained_jobs
+ end
+
+ test "a newer pause or terminal transition clears an undispatched admission", ctx do
+ for disposition <- [:new_pause, :complete, :failed] do
+ paused = paused_run(ctx)
+
+ trigger =
+ abort_trigger!(
+ "oban_jobs",
+ "json_extract(NEW.meta, '$.workflow_run') = '#{paused.run_id}'"
+ )
+
+ assert {:ok, receipt} = resume(paused)
+ assert Run.get(paused.run_id).budget_resume_pending == paused.budget_pause_id
+
+ retained =
+ case disposition do
+ :new_pause -> Run.budget_pause(paused.run_id, "new fixture pause", ["fixture"])
+ terminal -> finish(paused, terminal)
+ end
+
+ assert retained.budget_resume_pending == nil
+
+ if disposition == :new_pause do
+ assert retained.budget_pause_id != paused.budget_pause_id
+ assert retained.status == "budget_paused"
+ else
+ assert_clear_admission(retained)
+ end
+
+ drop_trigger!(trigger)
+ assert {:ok, ^receipt} = resume(paused)
+ assert Run.get(paused.run_id) == retained
+ assert jobs(paused) == []
+ assert length(admissions(paused)) == 1
+ assert length(events(paused)) == 1
+ end
+ end
+
+ defp paused_run(ctx, opts \\ []) do
+ run =
+ Run.start(
+ uid("budget-resume-run"),
+ ctx.name,
+ ctx.repo,
+ :fixture,
+ %{"working_dir" => ctx.workspace, "purpose" => "synthetic fixture"},
+ Keyword.get(opts, :budget, 1.0)
+ )
+
+ if spend = opts[:spend], do: SpendLedger.record(Run.spend_agent_id(run.run_id), spend)
+ Run.budget_pause(run.run_id, "fixture rail", ["fixture"])
+ end
+
+ defp resume(run, opts \\ []),
+ do: Actions.resume_run(run.run_id, run.budget_pause_id, opts)
+
+ defp expected_receipt(run, budget) do
+ %{
+ schema_version: "custode.workflow_budget_resume.v1",
+ run_id: run.run_id,
+ pause_id: run.budget_pause_id,
+ decision: "resumed",
+ previous_budget_usd: run.budget_usd,
+ budget_usd: budget,
+ status: "running"
+ }
+ end
+
+ defp assert_clear_admission(run) do
+ assert run.budget_pause_id == nil
+ assert run.budget_paused_at == nil
+ assert run.budget_resume_pending == nil
+ end
+
+ defp finish(run, :complete), do: Run.complete(run.run_id)
+ defp finish(run, :failed), do: Run.fail(run.run_id, "fixture failure")
+
+ defp jobs(run) do
+ Repo.all(
+ from(j in Oban.Job,
+ where: j.worker == "Custode.Workflow.NodeJob",
+ where: fragment("json_extract(?, '$.workflow_run')", j.meta) == ^run.run_id,
+ order_by: [asc: j.id]
+ )
+ )
+ end
+
+ defp admissions(run),
+ do:
+ Repo.all(from(r in BudgetResume.Row, where: r.run_id == ^run.run_id, order_by: [asc: r.id]))
+
+ defp events(run) do
+ Repo.all(
+ from(f in Feed.Entry,
+ where: f.event == "workflow_resumed",
+ where: fragment("json_extract(?, '$.run')", f.entry) == ^run.run_id,
+ order_by: [asc: f.id]
+ )
+ )
+ end
+
+ defp snapshot(run), do: {Run.get(run.run_id), jobs(run), admissions(run), events(run)}
+
+ defp concurrently(operations) do
+ caller = self()
+
+ tasks =
+ Enum.map(operations, fn operation ->
+ Task.async(fn ->
+ send(caller, {:ready, self()})
+
+ receive do
+ :decide -> operation.()
+ end
+ end)
+ end)
+
+ for _ <- tasks, do: assert_receive({:ready, _pid}, 1_000)
+ for task <- tasks, do: send(task.pid, :decide)
+ Enum.map(tasks, &Task.await(&1, 10_000))
+ end
+
+ defp abort_trigger!(table, predicate) do
+ trigger = uid("budget_resume_abort") |> String.replace("-", "_")
+
+ Repo.query!("""
+ CREATE TRIGGER #{trigger} BEFORE INSERT ON #{table}
+ WHEN #{predicate}
+ BEGIN SELECT RAISE(ABORT, 'fixture budget resume write failure'); END
+ """)
+
+ on_exit(fn -> drop_trigger!(trigger) end)
+ trigger
+ end
+
+ defp drop_trigger!(trigger), do: Repo.query!("DROP TRIGGER IF EXISTS #{trigger}")
+end
diff --git a/test/custode/workflow_launch_test.exs b/test/custode/workflow_launch_test.exs
index 9049f7fc..7a5e3502 100644
--- a/test/custode/workflow_launch_test.exs
+++ b/test/custode/workflow_launch_test.exs
@@ -5,6 +5,7 @@ defmodule Custode.WorkflowLaunchTest do
import Ecto.Query, only: [from: 2]
alias Custode.Feed
+ alias Custode.Operator.Actions
alias Custode.Repo
alias Custode.SpendLedger
alias Custode.Workflow
@@ -174,8 +175,6 @@ defmodule Custode.WorkflowLaunchTest do
end
describe "exact launch decisions (#846)" do
- alias Custode.Operator.Actions
-
test "shared actions refuse every explicit nonhuman actor before mutation" do
workflow = register(fixed_workflow(uid("authority")))
{:ok, proposal} = Launch.propose(workflow.name, uid("repo"))
@@ -1000,31 +999,36 @@ defmodule Custode.WorkflowLaunchTest do
assert %DateTime{} = signal.raised_at
end
- test "raise_and_resume/1 doubles the rail, lets the run go on and clears the signal" do
+ test "exact-pause resume doubles the rail, lets the run go on and clears the signal" do
workflow = register(fanning_workflow(uid("railed")))
{:ok, proposal} = Launch.propose(workflow.name, "owner/repo", budget_usd: 1.0)
{:ok, run} = Launch.approve(proposal.id)
park!(run)
- assert {:ok, resumed} = Launch.raise_and_resume(run.run_id)
+ assert {:ok, resumed} =
+ Actions.resume_run(
+ run.run_id,
+ Run.get(run.run_id).budget_pause_id
+ )
+
assert resumed.status == "running"
assert resumed.budget_usd == 2.0
assert Enum.any?(jobs(run.run_id), &(&1.meta["node_name"] == "merge"))
assert workflow_signals() == []
end
- test "raise_and_resume/1 refuses a run that is not parked, and one that does not exist" do
+ test "exact-pause resume refuses a run that is not parked, and one that does not exist" do
workflow = register(fixed_workflow(uid("fixed")))
{:ok, run} = Runner.launch(workflow.name, "owner/repo", run_id: uid("live"))
- assert {:error, :not_paused} = Launch.raise_and_resume(run.run_id)
- assert {:error, :no_such_run} = Launch.raise_and_resume(uid("ghost"))
+ assert {:error, :not_paused} = Actions.resume_run(run.run_id, uid("pause"))
+
+ assert {:error, :no_such_run} =
+ Actions.resume_run(uid("ghost"), uid("pause"))
end
end
describe "the operator's ops (#447)" do
- alias Custode.Operator.Actions
-
test "run/4 approves a launch from the signal's own args" do
workflow = register(fixed_workflow(uid("fixed")))
{:ok, proposal} = Launch.propose(workflow.name, "owner/repo", budget_usd: 3.5)
@@ -1066,9 +1070,11 @@ defmodule Custode.WorkflowLaunchTest do
park!(run)
assert Actions.handles?(:resume_run)
- assert :ok = Actions.run(:resume_run, %{run: run.run_id}, %{}, via: :liveview)
+ args = %{run: run.run_id, pause: Run.get(run.run_id).budget_pause_id}
+ assert :ok = Actions.run(:resume_run, args, %{}, via: :liveview)
assert %{status: "running", budget_usd: 2.0} = Run.get(run.run_id)
- assert {:error, :not_paused} = Actions.run(:resume_run, %{run: run.run_id})
+ assert :ok = Actions.run(:resume_run, args)
+ assert {:error, :pause_required} = Actions.run(:resume_run, %{run: run.run_id})
end
end
diff --git a/test/custode_web/console_live_test.exs b/test/custode_web/console_live_test.exs
index 2fc2d063..b8db740d 100644
--- a/test/custode_web/console_live_test.exs
+++ b/test/custode_web/console_live_test.exs
@@ -2,13 +2,17 @@ defmodule CustodeWeb.ConsoleLiveTest do
use ExUnit.Case, async: false
import Custode.TestHelpers
+ import Ecto.Query, only: [from: 2]
import Phoenix.ConnTest
import Phoenix.LiveViewTest
alias Custode.Asks
alias Custode.Availability.Collectors.Claude, as: ClaudeUsage
alias Custode.GitHub.Cache
+ alias Custode.Operator.Actions
+ alias Custode.Repo
alias Custode.Workflow
+ alias Custode.Workflow.BudgetResume
alias Custode.Workflow.Launch
alias Custode.Workflow.Run
alias CustodeWeb.Console.Item
@@ -1123,6 +1127,111 @@ defmodule CustodeWeb.ConsoleLiveTest do
assert name == workflow.name
end
+ for previous_decision <- [:admitted, :superseded] do
+ @previous_decision previous_decision
+
+ test "a delayed #{@previous_decision} budget click cannot resume the pause now in the console",
+ %{
+ conn: conn
+ } do
+ workflow =
+ Workflow.new!(uid("console-budget"), [
+ %Workflow.Stage{
+ name: :fixture,
+ nodes: [%Workflow.Node{name: :fixture, prompt: "fixture only", schema: %{}}]
+ }
+ ])
+
+ put_env!(:extra_workflows, %{workflow.name => workflow})
+ run = Run.start(uid("console-budget-run"), workflow.name, uid("repo"), :fixture, %{}, 1.0)
+ paused = Run.budget_pause(run.run_id, "fixture first pause", ["fixture"])
+
+ on_exit(fn ->
+ Repo.delete_all(
+ from(j in Oban.Job,
+ where: fragment("json_extract(?, '$.workflow_run')", j.meta) == ^run.run_id
+ )
+ )
+
+ Repo.delete_all(from(r in Run.Row, where: r.run_id == ^run.run_id))
+
+ Repo.delete_all(
+ from(f in Custode.Feed.Entry,
+ where: fragment("json_extract(?, '$.run')", f.entry) == ^run.run_id
+ )
+ )
+ end)
+
+ {:ok, view, _html} = live(conn, Rail.subject_path("#{workflow.name} on #{run.repo}"))
+ button = "button[phx-value-op=resume_run]"
+ assert has_element?(view, ~s(#{button}[phx-value-run="#{run.run_id}"]))
+ assert has_element?(view, ~s(#{button}[phx-value-pause="#{paused.budget_pause_id}"]))
+
+ original_click = %{
+ "op" => "resume_run",
+ "run" => run.run_id,
+ "pause" => paused.budget_pause_id
+ }
+
+ case @previous_decision do
+ :admitted ->
+ assert {:ok, _receipt} = Actions.resume_run(run.run_id, paused.budget_pause_id)
+
+ :superseded ->
+ Run.unpause(run.run_id)
+ end
+
+ current = Run.budget_pause(run.run_id, "fixture later pause", ["fixture"])
+ refute current.budget_pause_id == paused.budget_pause_id
+ send(view.pid, {:feed_entry, %{}})
+ assert has_element?(view, ~s(#{button}[phx-value-pause="#{current.budget_pause_id}"]))
+ refute has_element?(view, ~s(#{button}[phx-value-pause="#{paused.budget_pause_id}"]))
+
+ decision_query = from(r in BudgetResume.Row, where: r.run_id == ^run.run_id)
+
+ job_query =
+ from(j in Oban.Job,
+ where: fragment("json_extract(?, '$.workflow_run')", j.meta) == ^run.run_id,
+ order_by: [asc: j.id]
+ )
+
+ retained_decisions = Repo.all(decision_query)
+ retained_jobs = Repo.all(job_query)
+ html = render_click(view, "op", original_click)
+
+ case @previous_decision do
+ :admitted -> assert html =~ "Resume approved for this pause"
+ :superseded -> assert html =~ "different budget pause now"
+ end
+
+ for missing <- [
+ %{"op" => "resume_run"},
+ %{"op" => "resume_run", "run" => run.run_id},
+ %{"op" => "resume_run", "run" => run.run_id, "pause" => ""}
+ ] do
+ assert render_click(view, "op", missing) =~ "No pause reference was supplied"
+ assert Run.get(run.run_id) == current
+ assert Repo.all(decision_query) == retained_decisions
+ assert Repo.all(job_query) == retained_jobs
+ end
+
+ assert render_click(view, "op", %{
+ "op" => "resume_run",
+ "pause" => current.budget_pause_id
+ }) =~ "This workflow run no longer exists"
+
+ assert Run.get(run.run_id) == current
+ assert Repo.all(decision_query) == retained_decisions
+ assert Repo.all(job_query) == retained_jobs
+
+ view |> element(button) |> render_click()
+ resumed = Run.get(run.run_id)
+ assert resumed.status == "running"
+ assert resumed.budget_usd == current.budget_usd * 2
+ assert length(Repo.all(decision_query)) == length(retained_decisions) + 1
+ end
+ end
+
describe "the item pane" do
# on the live fleet "main is red" arrived with nothing to click
test "a signal's evidence is a link when the subject has a repository" do
diff --git a/test/custode_web/inbox_live_test.exs b/test/custode_web/inbox_live_test.exs
index 0b5bd848..a11b5c09 100644
--- a/test/custode_web/inbox_live_test.exs
+++ b/test/custode_web/inbox_live_test.exs
@@ -6,6 +6,7 @@ defmodule CustodeWeb.InboxLiveTest do
import Phoenix.LiveViewTest
alias Custode.Asks
+ alias Custode.Operator.Actions
alias Custode.Sensors.CiStatus
alias Custode.Sensors.CiStatus.Infrastructure
alias Custode.Workflow
@@ -308,16 +309,57 @@ defmodule CustodeWeb.InboxLiveTest do
%{conn: conn} do
run = parked_run!(register_workflow!(), 1.0)
+ pause = Run.get(run.run_id).budget_pause_id
+ assert is_binary(pause)
+
{:ok, view, _html} = live(conn, "/inbox")
- html = view |> element("button[phx-click=resume_run]") |> render_click()
+ button = "button[phx-click=resume_run][phx-value-pause='#{pause}']"
+ assert has_element?(view, button)
+ html = view |> element(button) |> render_click()
resumed = Run.get(run.run_id)
assert resumed.status == "running"
assert resumed.budget_usd == 2.0
- assert html =~ "resumed on a raised rail"
+ assert html =~ "budget $1.00 → $2.00"
+ assert html =~ "Resume approved for this pause"
refute html =~ "is parked on its budget rail"
end
+ test "an admitted pause retry and missing reference cannot resume a later pause", %{
+ conn: conn
+ } do
+ run = parked_run!(register_workflow!(), 1.0)
+ stale = Run.get(run.run_id).budget_pause_id
+
+ assert {:ok, _receipt} = Actions.resume_run(run.run_id, stale, via: :liveview)
+ Run.budget_pause(run.run_id, "run budget rail hit: $2.40 of $2.00", ["merge"])
+ current = Run.get(run.run_id)
+ assert current.status == "budget_paused"
+ refute current.budget_pause_id == stale
+
+ {:ok, view, _html} = live(conn, "/inbox")
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id, "pause" => stale})
+ assert html =~ "Resume approved for this pause"
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id, "pause" => "unknown-pause"})
+ assert html =~ "different budget pause now"
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id})
+ assert html =~ "No pause reference was supplied"
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id, "pause" => ""})
+ assert html =~ "No pause reference was supplied"
+
+ html = render_click(view, "resume_run", %{"id" => uid("inbox-missing"), "pause" => stale})
+ assert html =~ "This workflow run no longer exists"
+
+ after_clicks = Run.get(run.run_id)
+ assert after_clicks.status == "budget_paused"
+ assert after_clicks.budget_usd == 2.0
+ assert after_clicks.budget_pause_id == current.budget_pause_id
+ end
+
test "the header chip counts a launch proposal on every page", %{conn: conn} do
workflow = register_workflow!()
{:ok, _proposal} = Launch.propose(workflow.name, "owner/repo")
diff --git a/test/custode_web/workflows_live_test.exs b/test/custode_web/workflows_live_test.exs
index f1627dbd..3badcac1 100644
--- a/test/custode_web/workflows_live_test.exs
+++ b/test/custode_web/workflows_live_test.exs
@@ -8,6 +8,7 @@ defmodule CustodeWeb.WorkflowsLiveTest do
import Phoenix.ConnTest
import Phoenix.LiveViewTest
+ alias Custode.Operator.Actions
alias Custode.Repo
alias Custode.Workflow
alias Custode.Workflow.Launch
@@ -142,13 +143,57 @@ defmodule CustodeWeb.WorkflowsLiveTest do
assert html =~ "Raise the rail and resume"
assert has_element?(view, "[data-stage-state='done'] .sr-only", "Done")
- resumed = view |> element("button[phx-click='resume_run']") |> render_click()
+ pause = Run.get(run.run_id).budget_pause_id
+ assert is_binary(pause)
+
+ resumed =
+ view
+ |> element("button[phx-click='resume_run'][phx-value-pause='#{pause}']")
+ |> render_click()
+
assert resumed =~ "running"
+ assert resumed =~ "Resume approved for this pause"
# the rail was RAISED, not removed -- resuming onto the same ceiling would
# park the run again immediately
assert Run.get(run.run_id).budget_usd == 2.0
end
+ test "an admitted pause retry and missing reference cannot resume a later pause", %{conn: conn} do
+ workflow = register(uid("toy"))
+ {:ok, run} = Runner.launch(workflow.name, "owner/repo", budget_usd: 1.0)
+ Run.budget_pause(run.run_id, "run budget rail hit: $1.20 of $1.00", ["merge"])
+ stale = Run.get(run.run_id).budget_pause_id
+
+ assert {:ok, _receipt} =
+ Actions.resume_run(run.run_id, stale, via: :liveview)
+
+ Run.budget_pause(run.run_id, "run budget rail hit: $2.40 of $2.00", ["merge"])
+ current = Run.get(run.run_id)
+ refute current.budget_pause_id == stale
+
+ {:ok, view, _html} = live(conn, "/workflows")
+ card = "[data-workflow-run='#{run.run_id}']"
+
+ assert has_element?(
+ view,
+ card <> " button[phx-value-pause='#{current.budget_pause_id}']"
+ )
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id, "pause" => stale})
+ assert html =~ "Resume approved for this pause"
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id, "pause" => "unknown-pause"})
+ assert html =~ "different budget pause now"
+
+ html = render_click(view, "resume_run", %{"id" => run.run_id})
+ assert html =~ "No pause reference was supplied"
+
+ after_clicks = Run.get(run.run_id)
+ assert after_clicks.status == "budget_paused"
+ assert after_clicks.budget_usd == 2.0
+ assert after_clicks.budget_pause_id == current.budget_pause_id
+ end
+
test "a failed stage contains its full escaped error and keeps successful siblings", %{
conn: conn
} do
diff --git a/test/test_helper.exs b/test/test_helper.exs
index 539ba448..716c45fe 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -4,6 +4,7 @@
Custode.Repo.query!("UPDATE work_items SET parent_id = NULL")
Custode.Repo.query!("DELETE FROM workflow_runner_observations")
Custode.Repo.query!("DELETE FROM workflow_node_results")
+Custode.Repo.query!("DELETE FROM workflow_budget_resumes")
Custode.Repo.query!("DELETE FROM workflow_runs")
Custode.Repo.query!("UPDATE artifacts SET producer_attempt_id = NULL")
Custode.Repo.query!("UPDATE attempts SET caused_by_attempt_id = NULL")