fix(jobs): honor cancellation in terminal updates

Preserve cancel-requested runs as cancelled when late completion arrives, and keep their cancellation metadata intact.

Also cap synthetic progressive-message persistence with the caller's context so timeout handling stops promptly instead of hanging past deadlines.

Author: SamSaffron

cmd/progressive.go

@@ -19,9 +19,10 @@ const (
 	progressiveStopWhenDone    progressiveStopWhen = "done"
 	progressiveStopWhenTimeout progressiveStopWhen = "timeout"
 
-	progressiveDefaultFinalizeGrace = 5 * time.Minute
-	progressiveMaxFinalizeBudget    = 5 * time.Minute
-	progressiveMinFinalizeBudget    = 5 * time.Second
+	progressiveDefaultFinalizeGrace    = 5 * time.Minute
+	progressiveMaxFinalizeBudget       = 5 * time.Minute
+	progressiveMinFinalizeBudget       = 5 * time.Second
+	progressiveSyntheticMessageTimeout = 5 * time.Second
 )
 
 type askProgressiveOptions struct {
@@ -296,7 +297,10 @@ func runProgressiveSession(ctx context.Context, engine *llm.Engine, req llm.Requ
 		continueMsg := llm.UserText(expandProgressiveTemplate(opts.ContinueWith, mainCtx))
 		history = append(history, continueMsg)
 		if opts.OnSyntheticUserMessage != nil {
-			if err := opts.OnSyntheticUserMessage(context.Background(), continueMsg); err != nil {
+			msgCtx, cancel := progressiveSyntheticUserMessageContext(mainCtx)
+			err := opts.OnSyntheticUserMessage(msgCtx, continueMsg)
+			cancel()
+			if err != nil {
 				return buildProgressiveRunResult(opts.SessionID, exitReasonNatural, false, tracker.latest, lastText), err
 			}
 		}
@@ -425,7 +429,10 @@ func attemptProgressiveFinalization(parentCtx context.Context, engine *llm.Engin
 	finalPrompt := buildProgressiveFinalizePrompt(tracker.latest)
 	finalizeMsg := llm.UserText(finalPrompt)
 	if opts.OnSyntheticUserMessage != nil {
-		if err := opts.OnSyntheticUserMessage(context.Background(), finalizeMsg); err != nil {
+		msgCtx, cancel := progressiveSyntheticUserMessageContext(finalizeCtx)
+		err := opts.OnSyntheticUserMessage(msgCtx, finalizeMsg)
+		cancel()
+		if err != nil {
 			return false, ""
 		}
 	}
@@ -513,6 +520,15 @@ func progressiveHasRemainingBudget(parent context.Context, reserve time.Duration
 	return time.Until(deadline) > reserve
 }
 
+// progressiveSyntheticUserMessageContext caps synthetic user-message persistence
+// to a short window while still honoring the caller's cancellation/deadline.
+func progressiveSyntheticUserMessageContext(parent context.Context) (context.Context, context.CancelFunc) {
+	if parent == nil {
+		parent = context.Background()
+	}
+	return context.WithTimeout(parent, progressiveSyntheticMessageTimeout)
+}
+
 func progressiveFinalizeReserve(total time.Duration) time.Duration {
 	if total <= 0 {
 		return 0

cmd/progressive_test.go

@@ -3,10 +3,12 @@ package cmd
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"testing"
 	"time"
 
 	"github.com/samsaffron/term-llm/internal/llm"
+	toolpkg "github.com/samsaffron/term-llm/internal/tools"
 )
 
 type progressiveTestTool struct{}
@@ -175,6 +177,87 @@ func TestRunProgressiveSessionTimeoutDoesNotStartDetachedFinalizationAfterDeadli
 	}
 }
 
+func TestRunProgressiveSessionSyntheticContinuationPersistenceHonorsContext(t *testing.T) {
+	provider := llm.NewMockProvider("mock").WithCapabilities(llm.Capabilities{
+		ToolCalls:          true,
+		SupportsToolChoice: true,
+	})
+	provider.AddToolCall("progress-1", "update_progress", map[string]any{
+		"state": map[string]any{
+			"step": "draft",
+		},
+		"reason":  "milestone",
+		"message": "draft saved",
+	})
+	provider.AddTextResponse("draft answer")
+
+	engine := llm.NewEngine(provider, nil)
+	ctx, cancel := context.WithTimeout(context.Background(), 5050*time.Millisecond)
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := runProgressiveSession(ctx, engine, llm.Request{
+			Messages: []llm.Message{llm.UserText("Investigate X")},
+			MaxTurns: 8,
+			ToolChoice: llm.ToolChoice{
+				Mode: llm.ToolChoiceAuto,
+			},
+		}, progressiveRunOptions{
+			StopWhen: progressiveStopWhenTimeout,
+			OnSyntheticUserMessage: func(ctx context.Context, msg llm.Message) error {
+				<-ctx.Done()
+				return ctx.Err()
+			},
+		})
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("runProgressiveSession error = %v, want deadline exceeded", err)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("synthetic continuation persistence did not stop when its context expired")
+	}
+}
+
+func TestAttemptProgressiveFinalizationSyntheticPersistenceHonorsContext(t *testing.T) {
+	provider := llm.NewMockProvider("mock")
+	engine := llm.NewEngine(provider, nil)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		finalized, text := attemptProgressiveFinalization(ctx, engine, toolpkg.NewFinalizeProgressTool(), llm.Request{
+			Messages: []llm.Message{llm.UserText("Investigate X")},
+		}, nil, progressiveRunOptions{
+			OnSyntheticUserMessage: func(ctx context.Context, msg llm.Message) error {
+				<-ctx.Done()
+				return ctx.Err()
+			},
+		}, newProgressTracker(), 20*time.Millisecond, exitReasonTimeout)
+		if finalized {
+			t.Errorf("finalized = true, want false")
+		}
+		if text != "" {
+			t.Errorf("text = %q, want empty", text)
+		}
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		if len(provider.Requests) != 0 {
+			t.Fatalf("provider saw %d requests, want synthetic persistence failure to stop finalization before the model call", len(provider.Requests))
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("synthetic finalization persistence did not stop when its context expired")
+	}
+}
+
 func TestProgressiveFinalizationContextNaturalCompletionDetachesFromParent(t *testing.T) {
 	parent, cancelParent := context.WithCancel(context.Background())
 	finalizeCtx, finalizeCancel := progressiveFinalizationContext(parent, time.Second, exitReasonNatural)

cmd/serve_jobs_v2.go

@@ -1365,13 +1365,32 @@ func (m *jobsV2Manager) finishRun(runID string, status jobsV2RunStatus, result j
 	if runErr != nil {
 		errText = runErr.Error()
 	}
-	_, err := m.db.Exec(`UPDATE job_runs_v2 SET status = ?, finished_at = ?, exit_code = ?, error = ?, stdout = ?, stderr = ?, thinking = ?, response = ?, session_id = ?, exit_reason = ?, truncated = ?, turn_count = ?, input_tokens = ?, output_tokens = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
-		status, now, result.ExitCode, errText, result.Stdout, result.Stderr, result.Thinking, result.Response, result.SessionID,
-		exitReason, boolToInt(truncated), result.TurnCount, result.InputTokens, result.OutputTokens,
-		runID)
+	cancelledErrText := context.Canceled.Error()
+	res, err := m.db.Exec(`UPDATE job_runs_v2 SET status = CASE WHEN status = ? THEN ? ELSE ? END, finished_at = ?, exit_code = ?, error = CASE WHEN status = ? THEN ? ELSE ? END, stdout = ?, stderr = ?, thinking = ?, response = ?, session_id = ?, exit_reason = CASE WHEN status = ? THEN ? ELSE ? END, truncated = ?, turn_count = ?, input_tokens = ?, output_tokens = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status IN (?, ?, ?, ?)`,
+		jobsV2RunCancelRequested, jobsV2RunCancelled, status,
+		now, result.ExitCode,
+		jobsV2RunCancelRequested, cancelledErrText, errText,
+		result.Stdout, result.Stderr, result.Thinking, result.Response, result.SessionID,
+		jobsV2RunCancelRequested, exitReasonCancelled, exitReason,
+		boolToInt(truncated), result.TurnCount, result.InputTokens, result.OutputTokens,
+		runID, jobsV2RunQueued, jobsV2RunClaimed, jobsV2RunRunning, jobsV2RunCancelRequested)
 	if err != nil {
 		return err
 	}
+	affected, _ := res.RowsAffected()
+	if affected == 0 {
+		return nil
+	}
+
+	run, err := m.GetRun(runID)
+	if err != nil {
+		return err
+	}
+	status = run.Status
+	exitReason = run.ExitReason
+	truncated = run.Truncated
+	errText = run.Error
+
 	_ = m.addRunEvent(runID, string(status), "run finished", map[string]any{
 		"status":        status,
 		"attempt":       attempt,
@@ -1387,10 +1406,6 @@ func (m *jobsV2Manager) finishRun(runID string, status jobsV2RunStatus, result j
 	m.notifyRunDone(runID, status, result, exitReason, truncated, errText)
 
 	if status == jobsV2RunFailed || status == jobsV2RunTimedOut {
-		run, err := m.GetRun(runID)
-		if err != nil {
-			return nil
-		}
 		job, err := m.GetJob(run.JobID)
 		if err != nil {
 			return nil

cmd/serve_jobs_v2_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"path/filepath"
@@ -303,6 +304,81 @@ func TestJobsV2ManualTriggerAndCancel(t *testing.T) {
 	}
 }
 
+func TestJobsV2FinishRunDoesNotOverrideCancelRequested(t *testing.T) {
+	mgr, err := newJobsV2Manager(":memory:", 0, nil)
+	if err != nil {
+		t.Fatalf("newJobsV2Manager failed: %v", err)
+	}
+	defer func() { _ = mgr.Close() }()
+
+	job, err := mgr.CreateJob(jobsV2Job{
+		Name:          "cancel-authoritative",
+		Enabled:       true,
+		RunnerType:    jobsV2RunnerProgram,
+		RunnerConfig:  json.RawMessage(`{"command":"echo","args":["x"]}`),
+		TriggerType:   jobsV2TriggerManual,
+		TriggerConfig: json.RawMessage(`{}`),
+		RetryPolicy:   json.RawMessage(`{"max_attempts":2}`),
+	})
+	if err != nil {
+		t.Fatalf("CreateJob failed: %v", err)
+	}
+	if job.ID == "" {
+		t.Fatal("expected created job to have an id")
+	}
+
+	run, err := mgr.TriggerJob(job.ID)
+	if err != nil {
+		t.Fatalf("TriggerJob failed: %v", err)
+	}
+
+	started := time.Now().UTC()
+	if _, err := mgr.db.Exec(`UPDATE job_runs_v2 SET status = ?, started_at = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`, jobsV2RunRunning, started, run.ID); err != nil {
+		t.Fatalf("mark run running: %v", err)
+	}
+
+	cancelled, err := mgr.CancelRun(run.ID)
+	if err != nil {
+		t.Fatalf("CancelRun failed: %v", err)
+	}
+	if cancelled.Status != jobsV2RunCancelRequested {
+		t.Fatalf("cancelled status = %s, want %s", cancelled.Status, jobsV2RunCancelRequested)
+	}
+
+	result := jobsV2RunResult{Response: "partial output", ExitCode: 17}
+	if err := mgr.finishRun(run.ID, jobsV2RunFailed, result, errors.New("runner failed after cancel request"), run.Attempt); err != nil {
+		t.Fatalf("finishRun failed: %v", err)
+	}
+
+	current, err := mgr.GetRun(run.ID)
+	if err != nil {
+		t.Fatalf("GetRun failed: %v", err)
+	}
+	if current.Status != jobsV2RunCancelled {
+		t.Fatalf("run status = %s, want %s", current.Status, jobsV2RunCancelled)
+	}
+	if current.ExitReason != exitReasonCancelled {
+		t.Fatalf("exit reason = %q, want %q", current.ExitReason, exitReasonCancelled)
+	}
+	if current.Error != context.Canceled.Error() {
+		t.Fatalf("error = %q, want %q", current.Error, context.Canceled.Error())
+	}
+	if current.Response != result.Response {
+		t.Fatalf("response = %q, want %q", current.Response, result.Response)
+	}
+	if current.FinishedAt == nil {
+		t.Fatal("expected cancelled run to have finished_at set")
+	}
+
+	var retryRuns int
+	if err := mgr.db.QueryRow(`SELECT COUNT(*) FROM job_runs_v2 WHERE job_id = ? AND id != ?`, job.ID, run.ID).Scan(&retryRuns); err != nil {
+		t.Fatalf("count retry runs: %v", err)
+	}
+	if retryRuns != 0 {
+		t.Fatalf("retry runs = %d, want 0", retryRuns)
+	}
+}
+
 func TestJobsV2CreateDefaultsEnabledWhenOmitted(t *testing.T) {
 	mgr, err := newJobsV2Manager(":memory:", 0, nil)
 	if err != nil {

cmd/serve_response_runs.go

@@ -127,6 +127,19 @@ func (r *responseRun) appendEvent(event string, payload map[string]any) error {
 func (r *responseRun) complete(payload map[string]any, usage llm.Usage, sessionUsage llm.Usage) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
+	if r.cancelRequested {
+		r.status = "cancelled"
+		r.errorType = ""
+		r.errorMessage = ""
+		r.cancel = nil
+		r.cancelRequested = false
+		if response := mapValue(payload["response"]); len(response) > 0 {
+			response["status"] = "cancelled"
+			delete(response, "usage")
+			delete(response, "session_usage")
+		}
+		return r.appendEventLocked("response.cancelled", payload, true)
+	}
 	r.status = "completed"
 	r.errorType = ""
 	r.errorMessage = ""

cmd/serve_response_runs_test.go

@@ -129,6 +129,62 @@ func TestResponseRunSubscriberDroppedWhenBufferFull(t *testing.T) {
 	}
 }
 
+func TestResponseRunCompleteHonorsPendingCancellation(t *testing.T) {
+	run := newResponseRun("resp_cancelled", "sess_test", "", "mock", time.Now().Unix(), func() {})
+
+	if !run.cancelRun() {
+		t.Fatal("expected cancelRun to succeed")
+	}
+
+	if err := run.complete(map[string]any{
+		"response": map[string]any{
+			"id":            run.id,
+			"object":        "response",
+			"created":       run.created,
+			"model":         run.model,
+			"status":        "completed",
+			"usage":         usagePayload(llm.Usage{InputTokens: 3, OutputTokens: 4}),
+			"session_usage": usagePayload(llm.Usage{InputTokens: 5, OutputTokens: 6}),
+		},
+	}, llm.Usage{InputTokens: 3, OutputTokens: 4}, llm.Usage{InputTokens: 5, OutputTokens: 6}); err != nil {
+		t.Fatalf("complete failed: %v", err)
+	}
+
+	run.mu.Lock()
+	defer run.mu.Unlock()
+
+	if run.status != "cancelled" {
+		t.Fatalf("run status = %q, want cancelled", run.status)
+	}
+	if run.cancelRequested {
+		t.Fatal("cancelRequested should be cleared after terminal transition")
+	}
+	if len(run.events) != 1 {
+		t.Fatalf("events = %d, want 1", len(run.events))
+	}
+	if run.events[0].Event != "response.cancelled" {
+		t.Fatalf("event = %q, want response.cancelled", run.events[0].Event)
+	}
+
+	var payload map[string]any
+	if err := json.Unmarshal(run.events[0].Data, &payload); err != nil {
+		t.Fatalf("unmarshal terminal payload: %v", err)
+	}
+	response, ok := payload["response"].(map[string]any)
+	if !ok {
+		t.Fatalf("response payload type = %T", payload["response"])
+	}
+	if response["status"] != "cancelled" {
+		t.Fatalf("response status = %v, want cancelled", response["status"])
+	}
+	if _, ok := response["usage"]; ok {
+		t.Fatalf("cancelled payload unexpectedly retained usage: %#v", response["usage"])
+	}
+	if _, ok := response["session_usage"]; ok {
+		t.Fatalf("cancelled payload unexpectedly retained session_usage: %#v", response["session_usage"])
+	}
+}
+
 func TestResponseRunCompactionKeepsReplayWindowInOrder(t *testing.T) {
 	run := newResponseRun("resp_compact", "sess_test", "", "mock", time.Now().Unix(), func() {})
 	run.maxRetainedEvents = 3