From 7b3ca7cb4d088185d4bf4ab367bbc4955f2ad231 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Wed, 21 Oct 2026 11:03:59 +0800 Subject: [PATCH 1/4] test: accept workflow run cancellation at the HTTP boundary Drive POST /workflows/:wfid/runs/:rid/cancel through the real router and PostgreSQL: the request records one database instant and touches nothing else, replays under either key return that instant, the initiator-or-admin rule and the run scope are enforced, terminal runs answer 409, concurrent requests record once, and the request written over HTTP is what the scheduler pass consumes to settle a pending run cancelled without starting. Refs ora-space/circus#403 Co-Authored-By: Claude Code --- integration/workflow_run_cancel_test.go | 396 ++++++++++++++++++++++++ 1 file changed, 396 insertions(+) create mode 100644 integration/workflow_run_cancel_test.go diff --git a/integration/workflow_run_cancel_test.go b/integration/workflow_run_cancel_test.go new file mode 100644 index 000000000..28c3903fd --- /dev/null +++ b/integration/workflow_run_cancel_test.go @@ -0,0 +1,396 @@ +package integration + +// Acceptance for the workflow-side public cancellation request (specs decisions/cloud/workflow +// 0-cloud-side-scheduler-owns-workflow-execution.md D4/D6; circus#75 acceptance items 2 and 6). +// +// The writer already had core-level coverage in internal/core/workflow_run_cancel_db_test.go; what +// was missing was the real HTTP boundary the run page actually calls. These cases drive the shipped +// route and check the obligations the agent-side integration tests check for the other run kind: +// +// - the request records one instant and nothing else, and the instant it answers with is the +// durable one; +// - every repeat — same key, a fresh key, or after the run settled — returns that instant and +// changes nothing; +// - only the run's initiator and the tenant's administrators may cancel a run carrying frozen +// executor bindings; +// - a run outside this tenant/workflow, a soft-deleted run and a terminal run are refused; +// - concurrent requests record one instant and one request-caused version bump; +// - the request written through HTTP is the fact the scheduler's real pass consumes, so the +// pending run settles cancelled without starting, provisioning or leaving work behind. + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" + + "github.com/wanglongan587/cloud/internal/core" +) + +func workflowRunCancelPath(tenant, workflow, run string) string { + return "/api/v1/tenants/" + tenant + "/workflows/" + workflow + "/runs/" + run + "/cancel" +} + +// stageWorkflowRun creates a real workflow and publishes its snapshot, then stages a run row at the +// storage boundary. A run created through POST …/runs is settled by the dev simulator before a +// caller could ever see it pending, and the endpoint under test does not create runs — it cancels +// them — so the row is seeded in the state the case is about and every write from then on goes +// through the shipped route. +func stageWorkflowRun(t *testing.T, f *fixture, key, status, owner string) (string, string) { + t.Helper() + wf := f.call("POST", f.path("/workflows"), core.Object{ + "name": "Cancel target " + key, "graph": runGraph(), + }, key+"-workflow", 200).O("resource") + wfid := wf.S("id") + snap := f.call("POST", f.path("/workflows/"+wfid+"/publish"), core.Object{}, key+"-publish", 200).O("resource") + runID := uuid.NewString() + _, err := f.store.Pool.Exec(`INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input,node_states,created_by) + VALUES($1,$2,$3,$4,$5,$6,'{}'::jsonb,'{}'::jsonb,$7)`, + runID, f.tid, wfid, snap.S("id"), "Staged run "+key, status, owner) + must(t, err) + return wfid, runID +} + +// cancelWorkflowRun issues the production cancellation request as the fixture's verified user and +// asserts the exact status and — for refusals — the stable public code the client branches on. +func (f *fixture) cancelWorkflowRun(wfid, runID, key string, want int, code string) core.Object { + f.t.Helper() + return f.cancelWorkflowRunAs(f.user, f.tid, wfid, runID, key, core.Object{}, want, code) +} + +func (f *fixture) cancelWorkflowRunAs(user core.Claims, tenant, wfid, runID, key string, body core.Object, want int, code string) core.Object { + f.t.Helper() + out, status, err := f.client.Call(context.Background(), "POST", + workflowRunCancelPath(tenant, wfid, runID), "gateway", + core.Claims{RegisteredClaims: jwt.RegisteredClaims{Subject: "gateway-a"}}, &user, key, body) + must(f.t, err) + if status != want { + f.t.Fatalf("POST workflow run/cancel: want %d got %d %v", want, status, out) + } + if code != "" && out.S("code") != code { + f.t.Fatalf("POST workflow run/cancel: want code %q got %v", code, out) + } + return out +} + +// workflowCancelInstant reads the recorded request exactly as stored, so "no request" stays +// distinguishable from "a request recorded at the zero time". +func (f *fixture) workflowCancelInstant(runID string) (time.Time, bool) { + f.t.Helper() + var at *time.Time + must(f.t, f.store.Pool.QueryRow(`SELECT cancel_requested_at FROM workflow_runs WHERE id=$1`, runID).Scan(&at)) + if at == nil { + return time.Time{}, false + } + return *at, true +} + +// workflowRunFacts is every column of the run except the ones a request is allowed to move. A +// request that writes anything else — status, timestamps, node states — changes this string. +func (f *fixture) workflowRunFacts(runID string) string { + f.t.Helper() + var facts string + must(f.t, f.store.Pool.QueryRow(`SELECT (to_jsonb(r) - ARRAY['cancel_requested_at','version','updated_at'])::text + FROM workflow_runs r WHERE id=$1`, runID).Scan(&facts)) + return facts +} + +// assertWorkflowCancelRecorded proves the caller's key produced a durable idempotency record whose +// stored response is the 202 the caller saw — the thing that makes a replay a replay rather than a +// second decision. +func (f *fixture) assertWorkflowCancelRecorded(uid, key, instant string) { + f.t.Helper() + var stored int + var response string + must(f.t, f.store.Pool.QueryRow(`SELECT status, response::text FROM idempotency_records + WHERE tenant_id=$1 AND user_id=$2 AND key=$3`, f.tid, uid, key).Scan(&stored, &response)) + if stored != 202 { + f.t.Fatalf("the recorded status must be the 202 the caller saw, got %d", stored) + } + var recorded core.Object + must(f.t, json.Unmarshal([]byte(response), &recorded)) + if recorded.S("cancelRequestedAt") != instant || len(recorded) != 1 { + f.t.Fatalf("the recorded response must be the cancelRequestedAt the caller saw, got %s", response) + } +} + +// bindWorkflowModelToRun stages one run-level executor binding at the storage boundary, the way the +// creation transaction writes it: the binding is what makes the initiator-or-admin rule apply to +// this run, and the fixture seeds it directly because the connection and credential behind it are +// created by the model API, which is not what these cases are about. +func (f *fixture) bindWorkflowModelToRun(runID, userID string) { + f.t.Helper() + credential, connection, binding := newTestID(), newTestID(), newTestID() + _, err := f.store.Pool.Exec(`INSERT INTO personal_model_credentials(id,user_id,ciphertext,key_id) + VALUES($1,$2,repeat('a',64),'cancel-test')`, credential, userID) + must(f.t, err) + _, err = f.store.Pool.Exec(`INSERT INTO personal_model_connections(id,user_id,credential_id,name,protocol,base_url,auth_mode,models,enabled) + VALUES($1,$2,$3,'Cancel test','openai-completions','https://models.example.invalid/v1','bearer','[{"id":"vendor/model-v1"}]'::jsonb,true)`, + connection, userID, credential) + must(f.t, err) + _, err = f.store.Pool.Exec(`INSERT INTO run_model_bindings(id,workflow_run_id,user_id,connection_id,connection_version,credential_id,connection_name,protocol,base_url,auth_mode,model,slot_name) + VALUES($1,$2,$3,$4,1,$5,'Cancel test','openai-completions','https://models.example.invalid/v1','bearer','{"id":"vendor/model-v1"}'::jsonb,'default')`, + binding, runID, userID, connection, credential) + must(f.t, err) +} + +// The first request. One accepted cancellation records the request, reports the instant it was +// recorded, and does nothing else: the run keeps its status, timestamps and node states, no +// Workspace or execution work exists, and no command or operation is derived. Replays — the same +// key and a fresh one — return that instant and do not write again. +func TestWorkflowRunCancelRecordsOnlyTheRequest(t *testing.T) { + f := setup(t) + wfid, runID := stageWorkflowRun(t, f, "records", "pending", f.uid) + + if at, ok := f.workflowCancelInstant(runID); ok { + t.Fatalf("precondition: the staged run carries no request, got %v", at) + } + factsBefore := f.workflowRunFacts(runID) + versionBefore := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID) + + out := f.cancelWorkflowRun(wfid, runID, "wf-cancel-1", 202, "") + // The response is the recorded instant alone, and it has to be the whole body. + if len(out) != 1 { + t.Fatalf("the response must carry cancelRequestedAt and nothing else, got %v", out) + } + reported, err := time.Parse(time.RFC3339, out.S("cancelRequestedAt")) + must(t, err) + durable, ok := f.workflowCancelInstant(runID) + if !ok || !reported.Equal(durable) { + t.Fatalf("cancelRequestedAt = %s, want the durable instant %v", reported, durable) + } + + // The run is untouched apart from the request and its version bump. + if after := f.workflowRunFacts(runID); after != factsBefore { + t.Fatalf("recording a request moved the run:\nbefore %s\nafter %s", factsBefore, after) + } + if got := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID); got != versionBefore+1 { + t.Fatalf("recording the request must bump the version exactly once, got %d from %d", got, versionBefore+1) + } + for _, q := range []struct { + what string + query string + }{ + {"workspaces", `SELECT count(*) FROM workspaces WHERE workflow_run_id=$1`}, + {"execution work", `SELECT count(*) FROM execution_work WHERE workflow_run_id=$1`}, + {"thread commands", `SELECT count(*) FROM thread_commands WHERE workflow_run_id=$1`}, + {"operations", `SELECT count(*) FROM operations o JOIN workspaces w ON w.id=o.workspace_id WHERE w.workflow_run_id=$1`}, + } { + if n := f.scalar(q.query, runID); n != 0 { + t.Fatalf("a recorded request produced %d %s", n, q.what) + } + } + f.assertWorkflowCancelRecorded(f.uid, "wf-cancel-1", out.S("cancelRequestedAt")) + + // The run resource reports the request, which is what lets the page show a pending cancellation. + if got := f.call("GET", f.path("/workflows/"+wfid+"/runs/"+runID), nil, "", 200).S("cancelRequestedAt"); got != out.S("cancelRequestedAt") { + t.Fatalf("the run resource reports cancelRequestedAt=%q; want %q", got, out.S("cancelRequestedAt")) + } + + // Replays return the recorded instant instead of writing a second one. + same := f.cancelWorkflowRun(wfid, runID, "wf-cancel-1", 202, "") + fresh := f.cancelWorkflowRun(wfid, runID, "wf-cancel-2", 202, "") + for i, again := range []core.Object{same, fresh} { + if again.S("cancelRequestedAt") != out.S("cancelRequestedAt") { + t.Fatalf("replay %d returned %q; want the recorded instant %q", i, again.S("cancelRequestedAt"), out.S("cancelRequestedAt")) + } + } + if got := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID); got != versionBefore+1 { + t.Fatalf("a replay wrote again: version %d, want %d", got, versionBefore+1) + } + f.assertWorkflowCancelRecorded(f.uid, "wf-cancel-2", out.S("cancelRequestedAt")) + + // The body is `{}` and nothing else: an unknown field is refused rather than ignored. + if refused := f.cancelWorkflowRunAs(f.user, f.tid, wfid, runID, "wf-cancel-unknown", core.Object{"reason": "stop"}, 400, "unknown_field"); refused.S("code") != "unknown_field" { + t.Fatalf("an unknown body field must be refused, got %v", refused) + } +} + +// Cancel-before-start is the whole cancellation: the request written through HTTP is the fact the +// scheduler's pass reads, the run settles cancelled without ever claiming, no Workspace or derived +// operation appears, and later passes do not touch the terminal row. +func TestWorkflowRunCancelSettlesThroughTheScheduler(t *testing.T) { + f := setup(t) + wfid, runID := stageWorkflowRun(t, f, "scheduler", "pending", f.uid) + instant := f.cancelWorkflowRun(wfid, runID, "wf-cancel-pass", 202, "").S("cancelRequestedAt") + + must(t, f.store.AdvanceWorkflowRunsOnce(context.Background())) + run := f.call("GET", f.path("/workflows/"+wfid+"/runs/"+runID), nil, "", 200) + if run.S("status") != "cancelled" { + t.Fatalf("run status = %q; want cancelled", run.S("status")) + } + if run.S("startedAt") != "" { + t.Fatalf("a run cancelled before its claim stamped startedAt %q", run.S("startedAt")) + } + if run.S("finishedAt") == "" { + t.Fatal("a cancelled run carries no finishedAt") + } + if run.S("cancelRequestedAt") != instant { + t.Fatalf("the settlement cleared the recorded request: %q; want %q", run.S("cancelRequestedAt"), instant) + } + if states := run.O("nodeStates"); len(states) != 0 { + t.Fatalf("a cancelled-before-start run has node states %v; want none", states) + } + if n := f.scalar(`SELECT count(*) FROM workspaces WHERE workflow_run_id=$1`, runID); n != 0 { + t.Fatalf("a cancelled-before-start run left %d workspaces", n) + } + if n := f.scalar(`SELECT count(*) FROM execution_work WHERE workflow_run_id=$1`, runID); n != 0 { + t.Fatalf("a cancelled-before-start run left %d execution work items", n) + } + if n := f.scalar(`SELECT count(*) FROM operations o JOIN workspaces w ON w.id=o.workspace_id WHERE w.workflow_run_id=$1 AND o.kind='delete_workspace'`, runID); n != 0 { + t.Fatalf("a cancelled-before-start run derived %d Workspace deletions", n) + } + + // The request outlives the settlement: a repeat returns the same instant rather than a conflict. + if again := f.cancelWorkflowRun(wfid, runID, "wf-cancel-after-settlement", 202, ""); again.S("cancelRequestedAt") != instant { + t.Fatalf("a repeat after settlement returned %q; want %q", again.S("cancelRequestedAt"), instant) + } + version := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID) + must(t, f.store.AdvanceWorkflowRunsOnce(context.Background())) + if again := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID); again != version { + t.Fatalf("a later pass bumped the terminal run's version from %d to %d", version, again) + } +} + +// A run carrying frozen executor bindings is cancellable by its initiator and by the tenant's +// administrators only; an ordinary member who is neither is refused before learning whether the run +// was cancellable. A run with no binding keeps the member rule. +func TestWorkflowRunCancelAuthorizesTheInitiatorOrAnAdmin(t *testing.T) { + f := setup(t) + stranger := joinUser("wf-cancel-stranger") + callAs(t, f, stranger, "GET", "/api/v1/me", "", nil, 200) // account, no membership + member, memberID := f.addUser(t, "wf-cancel-member", "Member") + + // Bound to the fixture's administrator: the member is neither initiator nor admin. + boundByAdmin, boundByAdminRun := stageWorkflowRun(t, f, "bound-admin", "pending", f.uid) + f.bindWorkflowModelToRun(boundByAdminRun, f.uid) + f.cancelWorkflowRunAs(stranger, f.tid, boundByAdmin, boundByAdminRun, "wf-cancel-stranger", core.Object{}, 403, "membership_required") + f.cancelWorkflowRunAs(member, f.tid, boundByAdmin, boundByAdminRun, "wf-cancel-member-refused", core.Object{}, 403, "workflow_run_cancel_forbidden") + if _, recorded := f.workflowCancelInstant(boundByAdminRun); recorded { + t.Fatal("a refused caller must not record a request") + } + f.cancelWorkflowRunAs(f.user, f.tid, boundByAdmin, boundByAdminRun, "wf-cancel-admin", core.Object{}, 202, "") + + // Bound to the member: the initiator may cancel. + boundByMember, boundByMemberRun := stageWorkflowRun(t, f, "bound-member", "pending", memberID) + f.bindWorkflowModelToRun(boundByMemberRun, memberID) + f.cancelWorkflowRunAs(member, f.tid, boundByMember, boundByMemberRun, "wf-cancel-initiator", core.Object{}, 202, "") + + // No binding, no run-level rule: an active member may cancel. + unbound, unboundRun := stageWorkflowRun(t, f, "unbound", "pending", f.uid) + f.cancelWorkflowRunAs(member, f.tid, unbound, unboundRun, "wf-cancel-unbound", core.Object{}, 202, "") +} + +// Scope is the workflow routes' answer, unchanged: a run that is not this tenant's, not this +// workflow's, not a live row, or not an id is the same not-found a caller gets for a run that does +// not exist, and every refusal writes nothing. +func TestWorkflowRunCancelIsNotFoundOutsideItsScope(t *testing.T) { + f := setup(t) + wfid, runID := stageWorkflowRun(t, f, "scope", "pending", f.uid) + otherWorkflow, _ := stageWorkflowRun(t, f, "scope-other", "pending", f.uid) + otherTenant := f.moreTenant() + + for _, tc := range []struct{ name, tenant, workflow, run string }{ + {"cross tenant", otherTenant, wfid, runID}, + {"cross workflow", f.tid, otherWorkflow, runID}, + {"unknown run", f.tid, wfid, newTestID()}, + {"run id is not an id", f.tid, wfid, "not-a-uuid"}, + } { + t.Run(tc.name, func(t *testing.T) { + out := f.cancelWorkflowRunAs(f.user, tc.tenant, tc.workflow, tc.run, "wf-cancel-scope-"+tc.name, core.Object{}, 404, "not_found") + if out.S("code") != "not_found" { + t.Fatalf("%s: want not_found, got %v", tc.name, out) + } + }) + } + + t.Run("a soft-deleted run is gone", func(t *testing.T) { + _, gone := stageWorkflowRun(t, f, "scope-deleted", "pending", f.uid) + _, err := f.store.Pool.Exec(`UPDATE workflow_runs SET deleted_at=now() WHERE id=$1`, gone) + must(t, err) + f.cancelWorkflowRun(f.workflowOf(gone), gone, "wf-cancel-deleted", 404, "not_found") + if _, recorded := f.workflowCancelInstant(gone); recorded { + t.Fatal("a soft-deleted run must not record a request") + } + }) +} + +// workflowOf reads the workflow a staged run belongs to, which the route needs to address it. +func (f *fixture) workflowOf(runID string) string { + f.t.Helper() + var wid string + must(f.t, f.store.Pool.QueryRow(`SELECT workflow_id FROM workflow_runs WHERE id=$1`, runID).Scan(&wid)) + return wid +} + +// A run that already settled, and the never-written awaiting_input state, cannot be cancelled: there +// is no live execution to stop, so the request is refused rather than recorded as if it had decided +// anything. +func TestWorkflowRunCancelAfterSettlementIsRefused(t *testing.T) { + f := setup(t) + for _, status := range []string{"succeeded", "failed", "cancelled", "awaiting_input"} { + t.Run(status, func(t *testing.T) { + wfid, runID := stageWorkflowRun(t, f, "terminal-"+status, status, f.uid) + out := f.cancelWorkflowRun(wfid, runID, "wf-cancel-terminal-"+status, 409, "run_not_cancellable") + if out.S("code") != "run_not_cancellable" { + t.Fatalf("a %s run's request: want run_not_cancellable, got %v", status, out) + } + if _, recorded := f.workflowCancelInstant(runID); recorded { + t.Fatal("a refused request must not record an instant") + } + }) + } +} + +// Two callers racing to cancel the same run still record one request: the transactions serialize, +// whichever runs second reads the instant the first recorded, and both answers carry it. The row +// keeps one instant and one request-caused version bump. +func TestWorkflowRunCancelRacesToOneRequest(t *testing.T) { + f := setup(t) + wfid, runID := stageWorkflowRun(t, f, "race", "pending", f.uid) + versionBefore := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID) + + type answer struct { + instant string + status int + err error + } + answers := make([]answer, 2) + start := make(chan struct{}) + var group sync.WaitGroup + for i, key := range []string{"wf-cancel-race-a", "wf-cancel-race-b"} { + group.Add(1) + go func(i int, key string) { + defer group.Done() + <-start + out, status, err := f.client.Call(context.Background(), "POST", + workflowRunCancelPath(f.tid, wfid, runID), "gateway", + core.Claims{RegisteredClaims: jwt.RegisteredClaims{Subject: "gateway-a"}}, &f.user, key, core.Object{}) + answers[i] = answer{out.S("cancelRequestedAt"), status, err} + }(i, key) + } + close(start) + group.Wait() + + for i, got := range answers { + if got.err != nil || got.status != 202 || got.instant == "" { + t.Fatalf("request %d: status=%d instant=%q err=%v; want 202 carrying the recorded instant", i, got.status, got.instant, got.err) + } + if got.instant != answers[0].instant { + t.Fatalf("request %d recorded %q; request 0 recorded %q", i, got.instant, answers[0].instant) + } + } + durable, ok := f.workflowCancelInstant(runID) + reported, err := time.Parse(time.RFC3339, answers[0].instant) + must(t, err) + if !ok || !reported.Equal(durable) { + t.Fatalf("the row carries %v; want the reported instant %s", durable, answers[0].instant) + } + if got := f.scalar(`SELECT version FROM workflow_runs WHERE id=$1`, runID); got != versionBefore+1 { + t.Fatalf("a race wrote the version %d times, got %d from %d", got-versionBefore, got, versionBefore) + } +} From 5f8c7056f7388ae0a8352ca2b7d8f2d32e79c060 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Wed, 21 Oct 2026 11:03:59 +0800 Subject: [PATCH 2/4] test: order workflow run cancellation against the scheduler and callback Race the cancellation writer against the scheduler pass, against a session terminal callback, and against the end-confirmation window expiring with a late callback. Each case asserts the complete set of permitted final rows: the run is never pulled back into execution, a committed result is kept, a late result survives on the execution row, the Workspace deletion is declared, and the recorded request is retained. Refs ora-space/circus#403 Co-Authored-By: Claude Code --- .../core/workflow_run_cancel_order_db_test.go | 284 ++++++++++++++++++ 1 file changed, 284 insertions(+) create mode 100644 internal/core/workflow_run_cancel_order_db_test.go diff --git a/internal/core/workflow_run_cancel_order_db_test.go b/internal/core/workflow_run_cancel_order_db_test.go new file mode 100644 index 000000000..49dc31723 --- /dev/null +++ b/internal/core/workflow_run_cancel_order_db_test.go @@ -0,0 +1,284 @@ +package core + +// Ordering evidence for workflow-run cancellation (circus#75 acceptance items 2 and 6, G-026; +// specs decisions/cloud/workflow 0-cloud-side-scheduler-owns-workflow-execution.md D4/D6). The +// neighboring workflow_run_cancel_db_test.go cases drive one writer at a time; these drive two at +// once and assert the complete set of permitted final rows, not the common one. +// +// Every writer and the scheduler pass share the store's one advisory transaction lock, so the +// interleavings are serialized rather than true simultaneous commits — which is exactly the property +// the ordering evidence is about: whichever commits first decides, the second observes it in +// committed state, and no combination leaves the run, the node and the session disagreeing. + +import ( + "context" + "database/sql" + "sync" + "testing" +) + +// commitSessionResult stores one session's terminal result the way the Node's callback does: the +// result transaction alone, without advancing. It returns its error instead of failing the test so a +// caller can race it against another writer; callers in the test goroutine turn the error back into +// a failure. +func commitSessionResult(store *Store, runID, execution, nodeService, reason string) error { + _, err := store.transact(context.Background(), func(tx *transaction) Object { + executionResult(tx, &ControlRequest{Body: Object{"operationId": runID, "executionId": execution, "result": Object{"outcome": "agent_session_ended", "reason": reason, "node": Object{"nodeId": nodeService}}, "sequence": int64(1), "event": "ended"}}, true) + return Object{} + }) + return err +} + +// prepareRunningCancelScene advances a seeded run into a live, dispatched session and returns the +// facts the ordering cases race over. +func prepareRunningCancelScene(t *testing.T, store *Store) (runID, tid, wid, nodeService, execution string) { + t.Helper() + runID, tid, _, _, _ = seedWorkflowSessionScene(t, store) + wid = workflowRunWorkflowID(t, store, runID) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance to claim the run: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + execution = dispatchSession(t, store, runID, "a", nodeService, spec) + return runID, tid, wid, nodeService, execution +} + +// The cancellation writer and the scheduler pass run concurrently. Whichever wins, the pass that +// observes the request releases exactly one end request for the live session, and the run stays +// running until that session reports — the request never settles a live execution from under its +// node. The session's own terminal result then settles the final row: run cancelled, node +// cancelled, Workspace deletion declared, request retained. +func TestWorkflowRunCancelRacesTheSchedulerPass(t *testing.T) { + store := dispatcherDB(t) + runID, tid, wid, nodeService, execution := prepareRunningCancelScene(t, store) + caller := cancelCaller(t, store, tid) + + var cancelErr error + var cancelStatus int + var instant string + start := make(chan struct{}) + var group sync.WaitGroup + group.Add(2) + go func() { + defer group.Done() + <-start + out, status, err := workflowRunCancelCall(t, store, tid, wid, runID, "wf-cancel-pass-race", caller) + cancelErr, cancelStatus, instant = err, status, out.S("cancelRequestedAt") + }() + go func() { + defer group.Done() + <-start + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + // The pass swallows per-run failures by design; a scan failure is the only one that + // reaches here, and it must not be hidden. + t.Errorf("scheduler pass: %v", err) + } + }() + close(start) + group.Wait() + + if cancelErr != nil || cancelStatus != 202 || instant == "" { + t.Fatalf("the request: status=%d instant=%q err=%v; want 202 carrying the recorded instant", cancelStatus, instant, cancelErr) + } + // If the pass ran first, the request is the only side of the race that happened; the next pass + // must react to it. If the pass ran second, this is a replay that releases nothing new. + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance after the race: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + if status != "running" || runErr != "" { + t.Fatalf("after the race the run is %q error %q states=%v; want the live session held", status, runErr, states) + } + commands := claimCommands(t, store) + if len(commands) != 1 { + t.Fatalf("the race released %d commands; want exactly the one end request: %v", len(commands), commands) + } + cmd := commands[0] + if cmd.S("kind") != "end_session" || cmd.O("body").S("reason") != "cancelled" || + cmd.S("runId") != runID || cmd.S("executionId") != execution || cmd.S("nodeId") != nodeService { + t.Fatalf("command = %v; want end_session/cancelled for %s/%s/%s", cmd, runID, execution, nodeService) + } + + reportSessionEnd(t, store, runID, execution, nodeService, "cancelled") + status, runErr, states, version := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if status != "cancelled" || runErr != "" { + t.Fatalf("final run = %q error %q; want cancelled", status, runErr) + } + if agent == nil || agent.S("status") != "cancelled" || agent.S("error") != "" || agent.O("output").S("reason") != "cancelled" { + t.Fatalf("final node a = %v; want cancelled with reason cancelled", states["a"]) + } + assertRunWorkspaceReleased(t, store, runID) + if recorded := readCancelRequestedAt(t, store, runID); !recorded.Valid { + t.Fatal("the final state lost the recorded request") + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance after the settlement: %v", err) + } + if _, _, _, again := readWorkflowRun(t, store, runID); again != version { + t.Fatalf("a pass after settlement bumped the final row's version from %d to %d", version, again) + } +} + +// A session's own terminal callback and the cancellation request race. Both orders are permitted, +// and each has one consistent final row: if the request commits while the run is still running, the +// run is cancelled and the node keeps the result that actually ran; if the callback's pass settles +// the run first, the request is refused as no-longer-cancellable and nothing is recorded. A request +// is never recorded on a succeeded run, and a cancelled run never loses the result. +func TestWorkflowRunCancelAndTheSessionCallbackRace(t *testing.T) { + // Each attempt gets its own schema: the scene seeds one personal-model default per schema, and the + // outcome branch is read from whichever ordering the shared advisory lock produced, so repeating + // covers both without assuming a particular winner. + for attempt := 0; attempt < 4; attempt++ { + store := dispatcherDB(t) + runID, tid, wid, nodeService, execution := prepareRunningCancelScene(t, store) + caller := cancelCaller(t, store, tid) + + var cancelOut Object + var cancelErr error + var callbackErr error + start := make(chan struct{}) + var group sync.WaitGroup + group.Add(2) + go func() { + defer group.Done() + <-start + cancelOut, _, cancelErr = workflowRunCancelCall(t, store, tid, wid, runID, "wf-cancel-callback-race", caller) + }() + go func() { + defer group.Done() + <-start + callbackErr = commitSessionResult(store, runID, execution, nodeService, "idle_timeout") + if callbackErr == nil { + callbackErr = store.AdvanceWorkflowRunsOnce(context.Background()) + } + }() + close(start) + group.Wait() + + if callbackErr != nil { + t.Fatalf("attempt %d: the session callback: %v", attempt, callbackErr) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if runErr != "" || agent == nil || agent.S("status") != "succeeded" || agent.O("output").S("reason") != "idle_timeout" { + t.Fatalf("attempt %d: final state run=%q error=%q states=%v; want the idle_timeout result kept", attempt, status, runErr, states) + } + recorded := readCancelRequestedAt(t, store, runID) + if cancelErr == nil { + // Accepted: the request was recorded while the run was non-terminal, so it decides the run. + if cancelOut.S("cancelRequestedAt") == "" || !recorded.Valid { + t.Fatalf("attempt %d: an accepted request reported %q but recorded %v", attempt, cancelOut.S("cancelRequestedAt"), recorded) + } + if status != "cancelled" { + t.Fatalf("attempt %d: the accepted request left the run %q; want cancelled", attempt, status) + } + } else { + // Refused: the callback had already settled the run, and a refusal records nothing. + if code := ErrorCode(cancelErr).Code; code != "run_not_cancellable" { + t.Fatalf("attempt %d: the request failed with %q; want run_not_cancellable", attempt, code) + } + if recorded.Valid { + t.Fatalf("attempt %d: a refused request recorded %v", attempt, recorded.Time) + } + if status != "succeeded" { + t.Fatalf("attempt %d: a refused request left the run %q; want succeeded", attempt, status) + } + } + assertRunWorkspaceReleased(t, store, runID) + if status == "cancelled" { + // The recorded instant is the seam the outcome is read from; a settled run keeps it. + if again, _, err := workflowRunCancelCall(t, store, tid, wid, runID, "wf-cancel-callback-replay", caller); err != nil || again.S("cancelRequestedAt") != cancelOut.S("cancelRequestedAt") { + t.Fatalf("attempt %d: a repeat after settlement: instant=%q err=%v; want the recorded %q", attempt, again.S("cancelRequestedAt"), err, cancelOut.S("cancelRequestedAt")) + } + } + } +} + +// The end confirmation window expiring and a late session callback race. The pass either settles the +// run cancelled and leaves the callback's result on the execution row, or reads the committed result +// first and keeps the node's own outcome; the run is cancelled either way. The node never goes back +// to running, the result is never discarded, and the terminal row is not rewritten afterwards. +func TestWorkflowRunCancelWindowAndALateCallbackRace(t *testing.T) { + for attempt := 0; attempt < 4; attempt++ { + store := dispatcherDB(t) + runID, tid, wid, nodeService, execution := prepareRunningCancelScene(t, store) + caller := cancelCaller(t, store, tid) + mustCancelWorkflowRun(t, store, tid, wid, runID, "wf-cancel-window-race", caller) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("attempt %d: advance after the request: %v", attempt, err) + } + var requests int + if err := store.Pool.QueryRow(`SELECT count(*) FROM thread_commands + WHERE workflow_run_id=$1 AND kind='end_session'`, runID).Scan(&requests); err != nil { + t.Fatalf("attempt %d: count the end request: %v", attempt, err) + } + if requests != 1 { + t.Fatalf("attempt %d: %d end requests; want the one end request", attempt, requests) + } + // The Node never confirms; age the one end request past the confirmation window on the + // database clock so the next pass is the cancellation's own settlement. + if _, err := store.Pool.Exec(`UPDATE thread_commands SET created_at=clock_timestamp()-interval '6 minutes' + WHERE workflow_run_id=$1`, runID); err != nil { + t.Fatalf("attempt %d: age the end request: %v", attempt, err) + } + + var callbackErr error + start := make(chan struct{}) + var group sync.WaitGroup + group.Add(2) + go func() { + defer group.Done() + <-start + callbackErr = commitSessionResult(store, runID, execution, nodeService, "idle_timeout") + }() + go func() { + defer group.Done() + <-start + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Errorf("attempt %d: scheduler pass: %v", attempt, err) + } + }() + close(start) + group.Wait() + if callbackErr != nil { + t.Fatalf("attempt %d: report the late callback: %v", attempt, callbackErr) + } + + status, runErr, states, version := readWorkflowRun(t, store, runID) + agent, _ := asWorkflowObject(states["a"]) + if status != "cancelled" || runErr != "" { + t.Fatalf("attempt %d: final run = %q error %q states=%v; want cancelled", attempt, status, runErr, states) + } + if agent == nil { + t.Fatalf("attempt %d: final node a is missing", attempt) + } + switch { + case agent.S("status") == "succeeded" && agent.O("output").S("reason") == "idle_timeout": + // The pass read the committed result: the node keeps what actually ran. + case agent.S("status") == "cancelled" && agent.O("output").S("reason") == "cancelled" && agent.S("error") == "": + // The pass settled first: the callback's result belongs to the execution row only. + default: + t.Fatalf("attempt %d: final node a = %v; want either the idle_timeout result or cancelled", attempt, states["a"]) + } + var result sql.NullString + if err := store.Pool.QueryRow(`SELECT result FROM node_executions + WHERE operation_id=$1 AND workflow_node_id='a' AND kind='agent_session'`, runID).Scan(&result); err != nil { + t.Fatalf("attempt %d: read the execution result: %v", attempt, err) + } + if !result.Valid { + t.Fatalf("attempt %d: the late result was discarded; want it kept on the execution row", attempt) + } + assertRunWorkspaceReleased(t, store, runID) + if !readCancelRequestedAt(t, store, runID).Valid { + t.Fatalf("attempt %d: the final state lost the recorded request", attempt) + } + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("attempt %d: advance after settlement: %v", attempt, err) + } + if _, _, _, again := readWorkflowRun(t, store, runID); again != version { + t.Fatalf("attempt %d: a pass after settlement bumped the final row's version from %d to %d", attempt, version, again) + } + } +} From ce9eac8e57eb0a1c88f4aa010f11d54a629f839f Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Wed, 21 Oct 2026 14:21:36 +0800 Subject: [PATCH 3/4] test: assert the S5 release fact when workflow cancellation settles Terminal settlement now records the Workspace's release state in the same transaction. The cancellation evidence checks that half of the row too: a cancelled-before-start run (no Workspace) reports not_required with no release instant, and every live-session ordering case reports releasing with no reason or released_at until the delete completes. Co-Authored-By: Claude Code --- integration/workflow_run_cancel_test.go | 11 +++++++++++ .../core/workflow_run_cancel_order_db_test.go | 15 +++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/integration/workflow_run_cancel_test.go b/integration/workflow_run_cancel_test.go index 28c3903fd..cfad20358 100644 --- a/integration/workflow_run_cancel_test.go +++ b/integration/workflow_run_cancel_test.go @@ -244,6 +244,17 @@ func TestWorkflowRunCancelSettlesThroughTheScheduler(t *testing.T) { if n := f.scalar(`SELECT count(*) FROM operations o JOIN workspaces w ON w.id=o.workspace_id WHERE w.workflow_run_id=$1 AND o.kind='delete_workspace'`, runID); n != 0 { t.Fatalf("a cancelled-before-start run derived %d Workspace deletions", n) } + // S5 records the release fact in the transaction that settles the run. This run never + // provisioned a Workspace, so the honest state is not_required — and no release instant is + // invented for a deletion Cloud never observed. + var releaseState, releaseErr string + var releasedAt *time.Time + must(t, f.store.Pool.QueryRow(`SELECT workspace_release_state, workspace_release_error, workspace_released_at + FROM workflow_runs WHERE id=$1`, runID).Scan(&releaseState, &releaseErr, &releasedAt)) + if releaseState != "not_required" || releaseErr != "" || releasedAt != nil { + t.Fatalf("cancelled-before-start release = state %q reason %q released_at %v; want not_required with no release instant", + releaseState, releaseErr, releasedAt) + } // The request outlives the settlement: a repeat returns the same instant rather than a conflict. if again := f.cancelWorkflowRun(wfid, runID, "wf-cancel-after-settlement", 202, ""); again.S("cancelRequestedAt") != instant { diff --git a/internal/core/workflow_run_cancel_order_db_test.go b/internal/core/workflow_run_cancel_order_db_test.go index 49dc31723..acf5c886c 100644 --- a/internal/core/workflow_run_cancel_order_db_test.go +++ b/internal/core/workflow_run_cancel_order_db_test.go @@ -29,6 +29,18 @@ func commitSessionResult(store *Store, runID, execution, nodeService, reason str return err } +// assertWorkflowRunReleaseDeclared proves S5's half of the terminal settlement: the transaction that +// settled the run also recorded the release intent for its live Workspace, and nothing claims the +// Workspace is gone before Cloud observed the delete complete. +func assertWorkflowRunReleaseDeclared(t *testing.T, store *Store, runID string) { + t.Helper() + release := readWorkflowRelease(t, store, runID) + if release.state != workflowReleaseReleasing || release.reason != "" || release.releasedAt.Valid { + t.Fatalf("terminal release = state %q reason %q released_at %v; want releasing with no reason and no released_at", + release.state, release.reason, release.releasedAt) + } +} + // prepareRunningCancelScene advances a seeded run into a live, dispatched session and returns the // facts the ordering cases race over. func prepareRunningCancelScene(t *testing.T, store *Store) (runID, tid, wid, nodeService, execution string) { @@ -110,6 +122,7 @@ func TestWorkflowRunCancelRacesTheSchedulerPass(t *testing.T) { t.Fatalf("final node a = %v; want cancelled with reason cancelled", states["a"]) } assertRunWorkspaceReleased(t, store, runID) + assertWorkflowRunReleaseDeclared(t, store, runID) if recorded := readCancelRequestedAt(t, store, runID); !recorded.Valid { t.Fatal("the final state lost the recorded request") } @@ -187,6 +200,7 @@ func TestWorkflowRunCancelAndTheSessionCallbackRace(t *testing.T) { } } assertRunWorkspaceReleased(t, store, runID) + assertWorkflowRunReleaseDeclared(t, store, runID) if status == "cancelled" { // The recorded instant is the seam the outcome is read from; a settled run keeps it. if again, _, err := workflowRunCancelCall(t, store, tid, wid, runID, "wf-cancel-callback-replay", caller); err != nil || again.S("cancelRequestedAt") != cancelOut.S("cancelRequestedAt") { @@ -271,6 +285,7 @@ func TestWorkflowRunCancelWindowAndALateCallbackRace(t *testing.T) { t.Fatalf("attempt %d: the late result was discarded; want it kept on the execution row", attempt) } assertRunWorkspaceReleased(t, store, runID) + assertWorkflowRunReleaseDeclared(t, store, runID) if !readCancelRequestedAt(t, store, runID).Valid { t.Fatalf("attempt %d: the final state lost the recorded request", attempt) } From bbfe02fa1af705f6a9c5e9c6691f8492d214d3a7 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Wed, 21 Oct 2026 14:21:38 +0800 Subject: [PATCH 4/4] test: log which ordering each workflow cancellation race attempt hit The two ordering loops permit either interleaving and asserted the complete set of allowed final rows, but did not say which one occurred. Logging the branch makes the evidence auditable after a CI run instead of inferred from timing. Co-Authored-By: Claude Code --- internal/core/workflow_run_cancel_order_db_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/internal/core/workflow_run_cancel_order_db_test.go b/internal/core/workflow_run_cancel_order_db_test.go index acf5c886c..cbb1e202c 100644 --- a/internal/core/workflow_run_cancel_order_db_test.go +++ b/internal/core/workflow_run_cancel_order_db_test.go @@ -187,6 +187,7 @@ func TestWorkflowRunCancelAndTheSessionCallbackRace(t *testing.T) { if status != "cancelled" { t.Fatalf("attempt %d: the accepted request left the run %q; want cancelled", attempt, status) } + t.Logf("attempt %d: the request committed first; the callback's result settled the node and the run is cancelled", attempt) } else { // Refused: the callback had already settled the run, and a refusal records nothing. if code := ErrorCode(cancelErr).Code; code != "run_not_cancellable" { @@ -198,6 +199,7 @@ func TestWorkflowRunCancelAndTheSessionCallbackRace(t *testing.T) { if status != "succeeded" { t.Fatalf("attempt %d: a refused request left the run %q; want succeeded", attempt, status) } + t.Logf("attempt %d: the callback's pass settled the run first; the request was refused", attempt) } assertRunWorkspaceReleased(t, store, runID) assertWorkflowRunReleaseDeclared(t, store, runID) @@ -271,8 +273,10 @@ func TestWorkflowRunCancelWindowAndALateCallbackRace(t *testing.T) { switch { case agent.S("status") == "succeeded" && agent.O("output").S("reason") == "idle_timeout": // The pass read the committed result: the node keeps what actually ran. + t.Logf("attempt %d: the pass read the committed result; node a kept idle_timeout", attempt) case agent.S("status") == "cancelled" && agent.O("output").S("reason") == "cancelled" && agent.S("error") == "": // The pass settled first: the callback's result belongs to the execution row only. + t.Logf("attempt %d: the pass settled first; the late result stayed on the execution row", attempt) default: t.Fatalf("attempt %d: final node a = %v; want either the idle_timeout result or cancelled", attempt, states["a"]) }