diff --git a/integration/workflow_run_cancel_test.go b/integration/workflow_run_cancel_test.go new file mode 100644 index 000000000..cfad20358 --- /dev/null +++ b/integration/workflow_run_cancel_test.go @@ -0,0 +1,407 @@ +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) + } + // 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 { + 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) + } +} 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..cbb1e202c --- /dev/null +++ b/internal/core/workflow_run_cancel_order_db_test.go @@ -0,0 +1,303 @@ +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 +} + +// 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) { + 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) + assertWorkflowRunReleaseDeclared(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) + } + 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" { + 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) + } + 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) + 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. + 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"]) + } + 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) + assertWorkflowRunReleaseDeclared(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) + } + } +}