From 96f8149e20c04f6c41348f6caf87f0f154bde105 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Mon, 19 Oct 2026 13:44:13 +0800 Subject: [PATCH 1/3] refactor(core): make the Node-unreachable judgement one function for both run kinds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit runWorkspaceNodeUnknown only ever read its window argument, so it does not need the Store receiver, and the binding it tests can cover both kinds of run: a Workspace is tied to its run by issue_run_id or by workflow_run_id, which 0039 makes mutually exclusive, so one comparison answers "is this Workspace bound to this run?" without the caller having to say which kind it holds. Nothing about the judgement changes — its polarity, its database clock and its fail-closed window stay exactly as documented — but it is now the one place the heartbeat evidence behind "the Node is unreachable" is read, so the workflow scheduler's session settlement can ask it instead of defining a second one. Co-Authored-By: Claude Code --- internal/core/agent_run_release.go | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/internal/core/agent_run_release.go b/internal/core/agent_run_release.go index 6da3ea814..596425224 100644 --- a/internal/core/agent_run_release.go +++ b/internal/core/agent_run_release.go @@ -151,14 +151,20 @@ func (s *Store) deliveryGivenUp(t *transaction, o Object) bool { LIMIT 1`, runID, s.DeliveryGiveUpAfter.Seconds()) != nil { return true } - return s.runWorkspaceNodeUnknown(t, runID, wid, s.DeliveryUnreachableAfter) + return runWorkspaceNodeUnknown(t, runID, wid, s.DeliveryUnreachableAfter) } -// runWorkspaceNodeUnknown is the release half's one authoritative judgement that a run Workspace's -// Node state is unknown, and the only place the heartbeat evidence behind it is read: the run is -// bound to this Workspace and no live sandbox of it carries a Node that is connected and was last -// seen inside window. Both D5's delivery give-up and D8's delete give-up ask this one function, so the -// two sides cannot drift into two definitions of "the Node is unreachable". +// runWorkspaceNodeUnknown is the one authoritative judgement that a run Workspace's Node state is +// unknown, and the only place the heartbeat evidence behind it is read: the Workspace is bound to this +// run and no live sandbox of it carries a Node that is connected and was last seen inside window. +// IssueRun D5's delivery give-up, IssueRun D8's delete give-up and the workflow scheduler's in-flight +// session settlement (workflowAgentSessionOutcome) all ask this one function, so none of them can +// drift into a second definition of "the Node is unreachable". +// +// The binding is whichever column ties the Workspace to its run — `issue_run_id` for an IssueRun, +// `workflow_run_id` for a WorkflowRun. The two are mutually exclusive (0039's XOR), so one comparison +// covers both kinds without the caller having to say which it holds, and a Workspace bound to a +// different run of either kind is still refused. // // The polarity is deliberately "unknown" rather than "reachable": a Workspace whose row is missing, is // not bound to this run, or whose id is unusable is *not* unknown. Failing the other way round would @@ -167,13 +173,17 @@ func (s *Store) deliveryGivenUp(t *transaction, o Object) bool { // window is the caller's configured limit and is judged on the database clock, so a Cloud process with // a skewed clock cannot abandon a Workspace another replica would still be waiting for. A window that // was never configured (zero or negative) fails closed as "not unknown". -func (s *Store) runWorkspaceNodeUnknown(t *transaction, runID, wid string, window time.Duration) bool { +// +// A Workspace with no live sandbox at all is unknown immediately, window or not: there is no Node left +// to report anything, which is the case force stop leaves behind when it terminates the sandbox and +// the executions it was hosting in one transaction. +func runWorkspaceNodeUnknown(t *transaction, runID, wid string, window time.Duration) bool { if window <= 0 || !validID(wid) { return false } return t.one(` SELECT w.id FROM workspaces w - WHERE w.id=$1 AND w.issue_run_id=$2 + WHERE w.id=$1 AND (w.issue_run_id=$2 OR w.workflow_run_id=$2) AND NOT EXISTS ( SELECT 1 FROM sandbox_instances sb JOIN node_instances n ON n.sandbox_instance_id = sb.id @@ -208,7 +218,7 @@ func (s *Store) releaseGivenUp(t *transaction, o Object) bool { if t.one(`SELECT 1 FROM sandbox_instances WHERE workspace_id=$1 AND terminated_at IS NULL`, wid) == nil { return false } - return s.runWorkspaceNodeUnknown(t, o.S("id"), wid, s.DeliveryUnreachableAfter) + return runWorkspaceNodeUnknown(t, o.S("id"), wid, s.DeliveryUnreachableAfter) } // giveUpRunWorkspaceRelease is D8's terminal transition, run on the caller-owned *transaction: an From 33afd2908718bf300ec4995f6f621cd920b1d383 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Mon, 19 Oct 2026 13:44:52 +0800 Subject: [PATCH 2/3] feat(core): settle an in-flight workflow session whose Node is gone MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A workflow run's agent node is committed `running` while its session executes, and the pass holds it until the Node reports. When the Node can no longer report — its heartbeat has been stale past the deployment's window, or its sandbox is gone, which is the state a force stop leaves behind — holding the run waits forever for evidence that cannot arrive: the run stays `running` and its Workspace is never released. The settlement now asks the judgement the IssueRun release paths already make (runWorkspaceNodeUnknown) with the same window, so "the Node is unreachable" has one definition across run kinds, and fails the node under Cloud's own code `node_unavailable` rather than borrowing a Node-reported session end reason: nothing reported anything here, and the two must stay distinguishable. No committed result exists for the node, so it is never recorded succeeded (invariant 5), and the run follows the existing failure propagation into its terminal state and Workspace deletion. The window is threaded from the pass (Store.DeliveryUnreachableAfter) instead of introducing a second limit; an unconfigured window fails closed and holds the session, exactly as it does for D5 and D8. Tests: both shapes of a gone Node, silence inside the window holding, an unconfigured window holding, and a late real result staying on the execution row without rewriting the terminal run. Co-Authored-By: Claude Code --- cmd/server/main.go | 9 +- internal/config/config.go | 10 +- internal/core/store.go | 5 + internal/core/workflow_run_session.go | 45 +++- internal/core/workflow_run_session_db_test.go | 200 ++++++++++++++++++ internal/core/workflow_scheduler.go | 15 +- 6 files changed, 268 insertions(+), 16 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 4cdb06f4c..35ac1b548 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -262,9 +262,12 @@ func run() (runErr error) { // Workflow advance loop: the Cloud-side scheduler (specs#99 D1 条件②/④, landing order step 3). // A bounded rescan of pending and running workflow runs, each advanced in its own short // transaction; the ready set is recomputed from committed state only, so the pass is idempotent - // and a replica that crashes mid-tick resumes exactly where the committed state says. In its - // first step it executes start/condition/output inline and settles runs honestly; agent nodes - // fail the run fast with an explicit reason until the execution wiring lands in the next step. + // and a replica that crashes mid-tick resumes exactly where the committed state says. start, + // condition and output execute inline; an agent node runs as a real session through the control + // plane and settles from that session's terminal evidence. The settlement reads the Node + // unreachability window the delivery loops above configure — the same `DeliveryUnreachableAfter` + // D5 and D8 judge, not a second limit — so an in-flight session whose Node is gone fails its node + // instead of holding the run in `running`, and an unconfigured window fails closed by holding it. { const workflowAdvanceInterval = 10 * time.Second syncGroup.Add(1) diff --git a/internal/config/config.go b/internal/config/config.go index 425fec7f3..c5b6f5715 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -61,10 +61,12 @@ type IssueRunsConfig struct { // succeed. Zero or negative selects the approved default (2 hours). DeliveryGiveUpAfter time.Duration `mapstructure:"delivery_give_up_after"` - // DeliveryUnreachableAfter is IssueRun D5's unreachability window: how long the run Workspace's - // Node may be unreachable before Cloud treats it as gone and closes the run out in the IssueRun D8 - // give-up path, without waiting out the full failure window. Zero or negative selects the - // approved default (30 minutes). + // DeliveryUnreachableAfter is the deployment's one unreachability window: how long a run + // Workspace's Node may be unreachable before Cloud treats it as gone. IssueRun D5 uses it to close + // a run out in the D8 give-up path without waiting out the full failure window, and the workflow + // scheduler uses the same window to settle an in-flight session whose Node is gone — one limit, so + // two passes cannot disagree about whether the same Node is still there. Zero or negative selects + // the approved default (30 minutes). DeliveryUnreachableAfter time.Duration `mapstructure:"delivery_unreachable_after"` } diff --git a/internal/core/store.go b/internal/core/store.go index 11034ce6a..24a281c40 100644 --- a/internal/core/store.go +++ b/internal/core/store.go @@ -157,6 +157,11 @@ type Store struct { // are deployment configuration values, not columns, and both are judged entirely on the database // clock. Zero (the zero-value Store) means that condition never gives up, so the scan is a no-op // rather than a zero-length window that would abandon every delivery the moment it was declared. + // + // DeliveryUnreachableAfter is the deployment's one tolerance for a silent Node, not a + // delivery-only limit: D8's delete give-up reuses it, and so does the workflow scheduler, which + // settles an in-flight session whose Node is gone with it. A second limit would let two passes + // disagree about whether the same Node is still there. DeliveryGiveUpAfter time.Duration DeliveryUnreachableAfter time.Duration diff --git a/internal/core/workflow_run_session.go b/internal/core/workflow_run_session.go index 1ab4366a8..90a456c39 100644 --- a/internal/core/workflow_run_session.go +++ b/internal/core/workflow_run_session.go @@ -13,6 +13,14 @@ import ( // dispatches within the deadline fails its node instead of hanging the run in `running`. const workflowSessionStartDeadline = 15 * time.Minute +// workflowSessionNodeUnavailable is the reason an in-flight session settles with when the Node that +// owns it is gone. It is Cloud's own code for the diagnosis Cloud made, and deliberately not one of the +// Node-reported session end reasons (`interrupted`, `agent_failed`): nothing reported anything here, +// and borrowing the Node's vocabulary would make Cloud's judgement indistinguishable from a Node's +// report. It is also the published operation failure code for an unreachable Node — the same fact, so +// the same name. +const workflowSessionNodeUnavailable = "node_unavailable" + // settleWorkflowSessionEnded accepts a WorkflowRun's session terminal event. A WorkflowRun has no // Thread, no `phase` and no Revision delivery: its scheduler's next pass reads this very execution's // result as the node's terminal evidence (workflowAgentSessionOutcome), which is why this @@ -52,7 +60,8 @@ func workflowRunningAgentNode(states map[string]workflowNodeState) (string, bool // workflowAgentSessionOutcome settles a committed running agent node. Terminal evidence (this // node's own session execution's result) commits the node's terminal state; a -// dispatched-but-unreported session is held (the Node owns it); a session that never dispatched +// dispatched-but-unreported session is held (the Node owns it) for as long as the Node can still +// report it; a session whose Node is gone settles failed, and a session that never dispatched // fails the node once the startup deadline passes. The run advances on the next tick after any // commit. // @@ -60,17 +69,43 @@ func workflowRunningAgentNode(states map[string]workflowNodeState) (string, bool // (D4 executes its agent nodes serially through one Workspace), so evidence read at run scope — // "the newest agent_session result for this run" — would let the second node inherit the first // node's result and be recorded succeeded without ever executing (invariant 5). -func workflowAgentSessionOutcome(t *transaction, runID string, version int64, nodeID string, states map[string]workflowNodeState) (map[string]workflowNodeState, bool) { +func workflowAgentSessionOutcome(t *transaction, runID string, version int64, nodeID string, states map[string]workflowNodeState, unreachableAfter time.Duration) (map[string]workflowNodeState, bool) { e := t.one(`SELECT result FROM node_executions WHERE operation_id=$1 AND workflow_node_id=$2 AND kind='agent_session' AND result IS NOT NULL`, runID, nodeID) if e != nil { return commitWorkflowAgentTerminal(t, runID, version, nodeID, states, e.O("result")) } - // Dispatched but not yet terminal: the Node owns the session until it reports. A successful - // initial turn ends the session — Cloud releases one EndSession command and waits for the - // terminal result, which stays the node's only evidence (D4). + // Dispatched but not yet terminal: the Node owns the session until it reports — and a session may + // legitimately run for hours, so elapsed time is not evidence of anything. The Node is. When the + // session's Node can no longer report, holding the run `running` waits forever for evidence that + // cannot arrive, so the node settles failed with Cloud's own reason (invariant 5's converse: no + // committed result, so never succeeded). The judgement and the window are the ones the IssueRun + // release paths already make (runWorkspaceNodeUnknown), so "the Node is unreachable" has one + // definition across run kinds, and the state a force stop leaves behind — sandbox terminated, the + // executions it hosted marked — settles here instead of hanging. + // + // The polarity stays the shared judgement's: a run with no Workspace bound to it is not "unknown" + // and the session is held, exactly as D5 and D8 refuse to read a lookup problem as an abandoned + // run. The session path cannot reach that state anyway — startWorkflowAgentSession creates the + // Workspace before it commits the node `running` — and the binding is never cleared, only + // soft-deleted, which the live-sandbox half of the judgement already covers. if e := t.one(`SELECT execution_id FROM node_executions WHERE operation_id=$1 AND workflow_node_id=$2 AND kind='agent_session'`, runID, nodeID); e != nil { + if w := t.one("SELECT id FROM workspaces WHERE workflow_run_id=$1", runID); w != nil && + runWorkspaceNodeUnknown(t, runID, w.S("id"), unreachableAfter) { + state := states[nodeID] + state.Status = "failed" + state.FinishedAt = nowRFC3339() + state.Error = workflowSessionNodeUnavailable + state.Output = Object{"reason": workflowSessionNodeUnavailable} + states[nodeID] = state + finishWorkflowRun(t, runID, version, "failed", workflowSessionNodeUnavailable, states) + return states, false + } + // The Node is still there. A successful initial turn ends the session — Cloud releases one + // EndSession command and waits for the terminal result, which stays the node's only evidence + // (D4); the confirmation window bounds that wait (D7). A session with no successful turn is + // held. return workflowSessionEndRequest(t, runID, version, nodeID, e.S("executionId"), states) } // Never dispatched: the startup deadline applies. A node committed running with no registered diff --git a/internal/core/workflow_run_session_db_test.go b/internal/core/workflow_run_session_db_test.go index 879c4579b..16bce9df6 100644 --- a/internal/core/workflow_run_session_db_test.go +++ b/internal/core/workflow_run_session_db_test.go @@ -813,3 +813,203 @@ func TestWorkflowLiveModelBindingRefusesAnotherSlotsBinding(t *testing.T) { t.Fatalf("another slot's binding for this execution: err=%v; want model_grant_revoked", err) } } + +// A session whose Node can no longer report settles instead of holding the run `running` forever. +// The two shapes below are one fact to the scheduler — the Node is gone — and it is the fact the +// IssueRun release paths already judge (runWorkspaceNodeUnknown): a heartbeat stale past the +// configured window, and the state a force stop leaves behind, its sandbox terminated and the +// executions it hosted marked in the same transaction. Neither is an end reason any Node reported, +// so the node fails under Cloud's own code and the run fails with it. +func TestWorkflowSessionNodeGoneSettlesTheNodeFailed(t *testing.T) { + for _, tc := range []struct { + name string + nodeGone func(t *testing.T, store *Store, wsID string) + }{ + { + name: "heartbeat stale past the window", + nodeGone: func(t *testing.T, store *Store, wsID string) { + if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '2 hours' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("age the Node heartbeat: %v", err) + } + }, + }, + { + // runtime_force_stop.go's settlement, shape for shape: the sandbox and the Node hosted on + // it end together, the executions it was hosting are marked as terminated by the stop, and + // the Workspace stops. + name: "force stop terminated the sandbox", + nodeGone: func(t *testing.T, store *Store, wsID string) { + if _, err := store.Pool.Exec(`UPDATE node_instances SET ended_at=clock_timestamp(),connection_state='ended',version=version+1 WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("end the Node: %v", err) + } + if _, err := store.Pool.Exec(`UPDATE sandbox_instances SET terminated_at=clock_timestamp(),observed_state='terminated' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("terminate the sandbox: %v", err) + } + if _, err := store.Pool.Exec(`INSERT INTO runtime_force_stops(id,tenant_id,workspace_id,actor_user_id,reason,state,control_epoch,runtime_generation) + SELECT gen_random_uuid(), w.tenant_id, w.id, w.creator_user_id, 'force stop', 'terminating', 0, w.runtime_generation FROM workspaces w WHERE w.id=$1`, wsID); err != nil { + t.Fatalf("register the force stop: %v", err) + } + if _, err := store.Pool.Exec(`UPDATE node_executions SET terminated_by_force_stop_id=(SELECT id FROM runtime_force_stops WHERE workspace_id=$1) WHERE workspace_id=$1 AND result IS NULL`, wsID); err != nil { + t.Fatalf("mark the hosted executions: %v", err) + } + if _, err := store.Pool.Exec(`UPDATE workspaces SET desired_state='stopped',observed_state='stopped',admission_open=false WHERE id=$1`, wsID); err != nil { + t.Fatalf("stop the Workspace: %v", err) + } + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + store := dispatcherDB(t) + store.DeliveryUnreachableAfter = 30 * time.Minute + runID, _, _, _, _ := seedWorkflowSessionScene(t, store) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + dispatchSession(t, store, runID, "a", nodeService, spec) + tc.nodeGone(t, store, wsID) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance after the Node is gone: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + if status != "failed" || runErr != workflowSessionNodeUnavailable { + t.Fatalf("run settled %q error %q states=%v; want failed/%s", status, runErr, states, workflowSessionNodeUnavailable) + } + agent, _ := asWorkflowObject(states["a"]) + if agent == nil || agent.S("status") != "failed" || agent.S("error") != workflowSessionNodeUnavailable { + t.Fatalf("node a = %v; want failed under %s", states["a"], workflowSessionNodeUnavailable) + } + if reason := agent.O("output").S("reason"); reason != workflowSessionNodeUnavailable { + t.Fatalf("node a output = %v; want reason %s", agent.O("output"), workflowSessionNodeUnavailable) + } + if agent.S("finishedAt") == "" { + t.Fatalf("node a = %v; a settled node carries finishedAt", states["a"]) + } + // The settlement is Cloud's own diagnosis of evidence that never arrived, never a success + // inferred from its absence (invariant 5), and the run is terminal like any other: its + // unregistered work is gone and its Workspace deletion is declared. + assertRunWorkspaceReleased(t, store, runID) + }) + } +} + +// The window is the boundary, not the silence itself. A session may legitimately run for hours, so a +// Node that has not been heard from for a few minutes is a Node whose session is still running and +// the pass holds it — settling on any silence would fail runs whose Nodes are merely quiet. The same +// run settles once the silence outlives the configured window. +func TestWorkflowSessionNodeSilenceWithinWindowHolds(t *testing.T) { + store := dispatcherDB(t) + store.DeliveryUnreachableAfter = time.Hour + runID, _, _, _, _ := seedWorkflowSessionScene(t, store) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + dispatchSession(t, store, runID, "a", nodeService, spec) + if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '5 minutes' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("age the Node heartbeat inside the window: %v", err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance within the window: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if status != "running" || runErr != "" || agent == nil || agent.S("status") != "running" { + t.Fatalf("five minutes of silence settled the run %q error %q states=%v; want it held", status, runErr, states) + } + // The same run, once the silence has outlived the window the deployment configured. + if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '2 hours' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatalf("age the Node heartbeat past the window: %v", err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance past the window: %v", err) + } + status, runErr, states, _ = readWorkflowRun(t, store, runID) + agent, _ = asWorkflowObject(states["a"]) + if status != "failed" || runErr != workflowSessionNodeUnavailable || agent == nil || agent.S("status") != "failed" { + t.Fatalf("past the window: run %q error %q states=%v; want failed/%s", status, runErr, states, workflowSessionNodeUnavailable) + } +} + +// An unconfigured window fails closed here exactly as it does in the IssueRun release paths: with no +// configured tolerance for a silent Node, Cloud holds the session rather than failing a node on a +// limit nobody set, and a deployment that never set the key loses only the settlement — the run stays +// in the `running` it was already in, with nothing half-written. cmd/server configures the window for +// every real deployment. +func TestWorkflowSessionUnconfiguredWindowHoldsTheNode(t *testing.T) { + store := dispatcherDB(t) // DeliveryUnreachableAfter is zero: the unconfigured deployment + runID, _, _, _, _ := seedWorkflowSessionScene(t, store) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + dispatchSession(t, store, runID, "a", nodeService, spec) + // The most conclusive loss there is: the sandbox is gone and the Node with it. + if _, err := store.Pool.Exec(`UPDATE node_instances SET ended_at=clock_timestamp(),connection_state='ended' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatal(err) + } + if _, err := store.Pool.Exec(`UPDATE sandbox_instances SET terminated_at=clock_timestamp(),observed_state='terminated' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatal(err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance with no window configured: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if status != "running" || runErr != "" || agent == nil || agent.S("status") != "running" { + t.Fatalf("an unconfigured window settled the run %q error %q states=%v; want it held", status, runErr, states) + } +} + +// A real terminal result that arrives after Cloud settled the node is kept, and changes nothing. The +// run is already terminal, so the receipt commits no state and no pass rewrites the run from it +// (invariant 3) — but the executed work stays on the execution row, because that row is the only +// record of what actually ran. Cloud refused to infer an outcome for a Node it could not hear; it +// does not discard the outcome when the Node does report. +func TestWorkflowSessionResultAfterTheNodeSettledIsKept(t *testing.T) { + store := dispatcherDB(t) + store.DeliveryUnreachableAfter = 30 * time.Minute + runID, _, _, _, _ := seedWorkflowSessionScene(t, store) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + execution := dispatchSession(t, store, runID, "a", nodeService, spec) + if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '2 hours' WHERE workspace_id=$1`, wsID); err != nil { + t.Fatal(err) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance past the window: %v", err) + } + status, runErr, states, version := readWorkflowRun(t, store, runID) + if status != "failed" || runErr != workflowSessionNodeUnavailable { + t.Fatalf("run settled %q error %q; want failed/%s", status, runErr, workflowSessionNodeUnavailable) + } + // The Node comes back and reports the session's real end, which Cloud never heard. + reportSessionEnd(t, store, runID, execution, nodeService, "idle_timeout") + afterStatus, afterErr, afterStates, afterVersion := readWorkflowRun(t, store, runID) + if afterStatus != status || afterErr != runErr { + t.Fatalf("the late result moved the run to %q error %q; want %q error %q", afterStatus, afterErr, status, runErr) + } + if afterVersion != version { + t.Fatalf("the late result bumped the version to %d; want %d", afterVersion, version) + } + if jsonText(afterStates) != jsonText(states) { + t.Fatalf("the late result rewrote the node states:\n got %v\nwant %v", afterStates, states) + } + var raw []byte + if err := store.Pool.QueryRow(`SELECT result FROM node_executions WHERE execution_id=$1`, execution).Scan(&raw); err != nil { + t.Fatalf("read the execution result: %v", err) + } + kept := Object{} + if err := json.Unmarshal(raw, &kept); err != nil { + t.Fatal(err) + } + if kept.S("outcome") != "agent_session_ended" || kept.S("reason") != "idle_timeout" { + t.Fatalf("execution result = %v; want the reported session end kept as evidence", kept) + } +} diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go index c3eefbe8a..772d39a81 100644 --- a/internal/core/workflow_scheduler.go +++ b/internal/core/workflow_scheduler.go @@ -80,6 +80,12 @@ func (s *Store) scanAdvancableWorkflowRuns(ctx context.Context, limit int) ([]st // deliberately not surfaced here — the transaction already rolls back and the run is retried on the // next tick — so the loop must not flood logs because one run is transiently stuck. Only a scan-level // failure cancels the pass. +// +// The pass carries `DeliveryUnreachableAfter` into each run's advance: it is the deployment's one +// tolerance for a silent Node, and the settlement of an in-flight session whose Node is gone judges +// that same window through the same shared check (runWorkspaceNodeUnknown) the IssueRun release +// paths use. An unconfigured window (zero or negative) fails closed there — a session is held rather +// than failed — so a deployment that never set the limit loses nothing it was promised. func (s *Store) AdvanceWorkflowRunsOnce(ctx context.Context) error { if ctx.Err() != nil { return ctx.Err() @@ -90,7 +96,7 @@ func (s *Store) AdvanceWorkflowRunsOnce(ctx context.Context) error { } for _, id := range ids { if _, err := s.transact(ctx, func(t *transaction) Object { - advanceWorkflowRun(t, id) + advanceWorkflowRun(t, id, s.DeliveryUnreachableAfter) return Object{} }); err != nil { _ = err @@ -463,8 +469,9 @@ func (p *workflowProjection) incomingEdges(nodeID string) []Object { // advanceWorkflowRun claims one pending run and advances it as far as committed state allows, in the // caller's transaction. It is idempotent: a run that lost a claim race, was advanced concurrently, or -// is already terminal is left untouched. -func advanceWorkflowRun(t *transaction, runID string) { +// is already terminal is left untouched. unreachableAfter is the deployment's Node-unreachability +// window, carried through to the session settlement (see AdvanceWorkflowRunsOnce). +func advanceWorkflowRun(t *transaction, runID string, unreachableAfter time.Duration) { run := t.one(` SELECT r.*, s.graph FROM workflow_runs r JOIN workflow_snapshots s ON s.id = r.snapshot_id @@ -513,7 +520,7 @@ func advanceWorkflowRun(t *transaction, runID string) { // run), and the next wave waits for the terminal commit. The pass returns after any outcome — // a settled node lets the next tick execute its downstream wave. if nodeID, running := workflowRunningAgentNode(states); running { - if _, cont := workflowAgentSessionOutcome(t, runID, version, nodeID, states); !cont { + if _, cont := workflowAgentSessionOutcome(t, runID, version, nodeID, states, unreachableAfter); !cont { return } // A settled session commits its terminal state and continues the pass; that commit bumped From db0cddd0d2c4112a75163dcae9c073efd6372956 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Mon, 19 Oct 2026 13:45:17 +0800 Subject: [PATCH 3/3] docs(workflows): record the Node-unreachability settlement of in-flight sessions The scheduling section named the reasons a session fails to start; add the one it fails with when its Node is gone, including the window it reads and the fact that `node_unavailable` is Cloud's own diagnosis rather than a session end reason a Node reported. The configuration table now says `delivery_unreachable_after` is also the "is the Node still there" window, since a deployment that sets it sets both. Co-Authored-By: Claude Code --- docs/operations/configuration.en.md | 2 +- docs/operations/configuration.md | 2 +- docs/workflows/workflows.en.md | 11 ++++++++++- docs/workflows/workflows.md | 6 +++++- 4 files changed, 17 insertions(+), 4 deletions(-) diff --git a/docs/operations/configuration.en.md b/docs/operations/configuration.en.md index 26cb17aad..ac97d01ec 100644 --- a/docs/operations/configuration.en.md +++ b/docs/operations/configuration.en.md @@ -50,7 +50,7 @@ Each process in this repository has its own configuration sources: | `issue_runs.thread_idle_timeout` | `15m` | The Thread idle window; non-positive values take the default | | `issue_runs.agent_start_timeout`, `agent_restore_timeout`, `agent_end_confirm_timeout` | `2m`, `6m`, `5m` | Startup windows; an explicit non-positive value stops startup, see [Bounded startup](../agent-runs/startup-settlement.en.md) | | `issue_runs.agent_workspace_blocked_timeout` | `30m` | The give-up window for a run Workspace creation stuck in `blocked`; validated like the startup windows | -| `issue_runs.delivery_give_up_after`, `delivery_unreachable_after` | `2h`, `30m` | Give-up limits for delivery and deletion; non-positive values take the defaults | +| `issue_runs.delivery_give_up_after`, `delivery_unreachable_after` | `2h`, `30m` | Give-up limits for delivery and deletion; `delivery_unreachable_after` is also the window that decides whether a Node is still there, and an in-flight workflow session settles on it; non-positive values take the defaults | | `object_store.*` | `enabled: false` | See [Revisions](../agent-runs/revisions.en.md#object-storage): `endpoint`, `public_endpoint`, `region`, `bucket`, `path_style`, `access_key_id_file`, `secret_access_key_file`, `upload_grant_ttl` | | `repository.enabled` | `false` | Enables the external Git line and the credential routes; with it off Projects are still created, but no sync runs and credential creation answers `503`; see [The external Git line](../repositories/external-git.en.md) | | `repository.executor_url`, `executor_ca_file`, `dispatcher_cert_file`, `dispatcher_key_file` | URL defaults to `https://repo-executor:8471`; files required | The Repo Executor's task listener and Cloud's dispatcher mTLS client identity | diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md index 172747eba..83399372a 100644 --- a/docs/operations/configuration.md +++ b/docs/operations/configuration.md @@ -45,7 +45,7 @@ | `issue_runs.thread_idle_timeout` | `15m` | Thread 空闲窗口；非正值取默认 | | `issue_runs.agent_start_timeout`、`agent_restore_timeout`、`agent_end_confirm_timeout` | `2m`、`6m`、`5m` | 启动窗口，显式配置非正值时拒绝启动，见 [有界启动](../agent-runs/startup-settlement.md) | | `issue_runs.agent_workspace_blocked_timeout` | `30m` | 运行 Workspace 创建停在 `blocked` 的放弃窗口，规则同上 | -| `issue_runs.delivery_give_up_after`、`delivery_unreachable_after` | `2h`、`30m` | 交付与删除的放弃上限；非正值取默认 | +| `issue_runs.delivery_give_up_after`、`delivery_unreachable_after` | `2h`、`30m` | 交付与删除的放弃上限；`delivery_unreachable_after` 同时是"Node 是否还在"的判定窗口，工作流运行里进行中的会话也按它结算；非正值取默认 | | `object_store.*` | `enabled: false` | 见 [Revision](../agent-runs/revisions.md#对象存储)：`endpoint`、`public_endpoint`、`region`、`bucket`、`path_style`、`access_key_id_file`、`secret_access_key_file`、`upload_grant_ttl` | | `repository.enabled` | `false` | 打开外部 Git 线与凭据路由；关闭时 Project 仍可创建，但没有同步，凭据创建返回 `503`，见 [外部 Git 线](../repositories/external-git.md) | | `repository.executor_url`、`executor_ca_file`、`dispatcher_cert_file`、`dispatcher_key_file` | URL 默认 `https://repo-executor:8471`，文件必填 | Repo Executor 的任务监听，以及 Cloud 调度方的 mTLS 客户端身份 | diff --git a/docs/workflows/workflows.en.md b/docs/workflows/workflows.en.md index 77a82cce0..5a0528ce4 100644 --- a/docs/workflows/workflows.en.md +++ b/docs/workflows/workflows.en.md @@ -108,7 +108,16 @@ seconds): bounded. A Workspace that cannot be created fails the node as `workspace_unavailable`; a session that finds its frozen plugin version or slot binding missing fails as `agent_plugin_version_unfrozen` or `workflow_executor_unconfigured` — never hanging in `running` or reporting success for work that never happened. - Once the run reaches a terminal state, Cloud declares the deletion of its Workspace. + **When a session has started but its Node can no longer report**, the node settles failed as `node_unavailable` + and the run reaches a terminal state: the judgement is the Node's reachability, not how long the session has been + running — a heartbeat older than `issue_runs.delivery_unreachable_after` (30 minutes by default), or a sandbox gone + together with the Node on it (exactly what a force stop leaves behind), both mean the Node is not there. That + judgement and its window are the ones the IssueRun delivery and release paths already use + (`runWorkspaceNodeUnknown`), so the two run kinds cannot answer "is the Node still there?" differently; an + unconfigured window does not degrade into a zero-length one, and the session is held rather than failed on the + spot. `node_unavailable` is Cloud's own diagnosis — it is also the published operation failure code for an + unreachable Node — and not a session end reason a Node reported. Once the run reaches a terminal state, Cloud + declares the deletion of its Workspace. - **A workflow session is single-turn, and Cloud decides when it ends.** Once the initial turn's `turnEnded` receipt arrives (with a `stop_reason` other than `cancelled`), the scheduler ends the session through the existing `EndSession` command relay; the session result the Node reports remains the node's only terminal evidence. If the diff --git a/docs/workflows/workflows.md b/docs/workflows/workflows.md index 11d9d4056..2b24d3b4d 100644 --- a/docs/workflows/workflows.md +++ b/docs/workflows/workflows.md @@ -76,7 +76,11 @@ WorkflowRun 由 Cloud server 里的调度循环推进（`AdvanceWorkflowRunsOnce - **Agent 节点在一次真实会话里执行。** 运行第一次调度到 Agent 节点时，Cloud 在运行冻结的 Project 里创建这次运行的 Workspace（isolated，属于该运行）， 等它有连上的 Node 后，按该节点引用的槽位取出创建时冻结的模型绑定，把会话工作项交给 Controller；会话的终态就是该节点的终态。同一运行同一时刻只有一个会话， 下一条边上的节点等它结束才开始，会话开始前的等待有上限。运行 Workspace 建不出来时节点以 `workspace_unavailable` 失败，会话开始时发现冻结的插件版本或槽位绑定 - 缺失时以 `agent_plugin_version_unfrozen`、`workflow_executor_unconfigured` 失败，都不会停在 `running` 或报告没有发生的成功。运行进入终态后，Cloud 声明删除 + 缺失时以 `agent_plugin_version_unfrozen`、`workflow_executor_unconfigured` 失败，都不会停在 `running` 或报告没有发生的成功。**会话已经开始、但它的 Node 再也报不出消息时**， + 节点以 `node_unavailable` 结算失败，运行随之进入终态：依据是 Node 的可达性，而不是会话已经跑了多久——心跳超过 `issue_runs.delivery_unreachable_after`（默认 30 分钟）， + 或者沙箱连同其上的 Node 一起消失（强制停止留下的正是这个状态），都表示 Node 已经不在了。这个判断和它的窗口与 IssueRun 交付、释放路径用的是同一个 + （`runWorkspaceNodeUnknown`），两种运行不会对"Node 还在吗"给出不同答案；窗口未配置时不会退化成零长度窗口，会话保持等待而不是被立刻误判为失败。 + `node_unavailable` 是 Cloud 自己下的判断（也是 Node 不可达时公开的操作失败码），不是 Node 上报的会话结束原因。运行进入终态后，Cloud 声明删除 这次运行的 Workspace。 - **工作流会话是单回合的，结束由 Cloud 决定。** 初始回合的 `turnEnded` 回执（`stop_reason` 不是 `cancelled`）到达后，调度器通过既有的 `EndSession` 命令让 Node 结束会话；Node 上报的会话执行结果仍是节点终态的唯一证据。结束请求放出后超过确认窗口仍没有会话终态时，节点以