fix(workers): close the four worker-observability gaps (#248)

internal/pipeline/implstage.go

@@ -411,13 +411,40 @@ func impRunRetryLoop(env impRetryEnv, plan ImplementationPlan, cfg StageConfig,
 				KillStaleProcessTree(s.Pid)
 			}
 		}()
-		log.Info("implement", fmt.Sprintf("Attempt %d finished", displayAttempt), []ledger.KV{
+		finished := []ledger.KV{
 			{Key: "exitCode", Value: result.ExitCode},
 			{Key: "timedOut", Value: result.TimedOut},
 			{Key: "durationMs", Value: result.DurationMs},
 			{Key: "events", Value: len(result.Events)},
-		})
+		}
+		if result.TimedOut {
+			// Issue #248: the kill cause belongs on the row — a
+			// productive wall-kill and a zero-event burn must not share
+			// an indistinguishable "timedOut true, events 0" line.
+			finished = append(finished,
+				ledger.KV{Key: "watchdogFired", Value: result.WatchdogFired},
+				ledger.KV{Key: "coldStart", Value: result.ColdStart},
+				ledger.KV{Key: "clockResets", Value: result.ClockResets},
+				ledger.KV{Key: "meaningfulBytes", Value: result.MeaningfulBytes},
+			)
+		}
+		log.Info("implement", fmt.Sprintf("Attempt %d finished", displayAttempt), finished)
 		if result.TimedOut || result.ExitCode != 0 {
+			if impIsProductiveWallKill(result) {
+				// Own classification (issue #248): the attempt streamed
+				// real work before the wall killed it. Retry semantics
+				// stay identical to the transient path — backoff, then
+				// the next dispatch continues the session via -c — but
+				// the row says what actually happened.
+				infraRetries++
+				data := []ledger.KV{
+					{Key: "clockResets", Value: result.ClockResets},
+					{Key: "meaningfulBytes", Value: result.MeaningfulBytes},
+				}
+				log.Warn("implement", fmt.Sprintf("wall-timeout (productive): attempt %d streamed %d clock resets / %d meaningful bytes before the kill; retrying with resume (infra retry %d)", displayAttempt, result.ClockResets, result.MeaningfulBytes, infraRetries), data)
+				time.Sleep(time.Duration(backoff(infraRetries)) * time.Millisecond)
+				continue
+			}
 			if impIsInfraTransient(result) {
 				infraRetries++
 				func() {
@@ -482,10 +509,44 @@ func impRunRetryLoop(env impRetryEnv, plan ImplementationPlan, cfg StageConfig,
 	return ImplementResult{OK: false, Worker: env.workerName, Attempts: attempts, WorktreePath: env.worktreePath, FailureClass: lastFailureClass}, false, nil
 }
 
-// impIsInfraTransient mirrors the isInfraTransient closure (Q31): a cold-start
-// kill is transient infra alongside the watchdog timeout — a wedged CLI init
-// is not the worker's fault.
+// impMeaningfulBytesThreshold: a wall-killed attempt counts as
+// productive when the stream classifier saw meaningful output above this
+// byte floor (Q33), or at least one classified clock reset. The
+// "did the worker actually do anything" bar from the issue #248
+// evidence run (a 72-minute productive attempt carried tool activity
+// and hundreds of KB of meaningful bytes; a wedged provider burn
+// streams thinking-only or nothing at all).
+const impMeaningfulBytesThreshold = 1024
+
+// impIsProductiveWallKill reports whether a wall/watchdog kill landed on
+// an attempt that demonstrably made progress (issue #248 required
+// change 3): meaningful stream bytes above threshold, or at least one
+// adapter-classified clock reset. Such a kill is NOT "transient infra"
+// — it is its own outcome, retried via the -c resume path but classified
+// and logged as what happened.
+func impIsProductiveWallKill(r workers.WorkerResult) bool {
+	if !r.TimedOut {
+		return false
+	}
+	if r.ColdStart {
+		return false
+	}
+	if r.ClockResets > 0 || r.MeaningfulBytes >= impMeaningfulBytesThreshold {
+		return true
+	}
+	return false
+}
+
+// impIsInfraTransient mirrors the isInfraTransient closure (Q31): a
+// cold-start kill is transient infra alongside a zero-progress watchdog
+// timeout — a wedged CLI init is not the worker's fault. A productive
+// wall-kill (progress present before the kill) is deliberately NOT
+// transient: it surfaces as its own "wall-timeout (productive)" row and
+// still retries through the same resume loop.
 func impIsInfraTransient(r workers.WorkerResult) bool {
+	if r.TimedOut && impIsProductiveWallKill(r) {
+		return false
+	}
 	if r.TimedOut || r.ColdStart {
 		return true
 	}

internal/pipeline/implstage_test.go

@@ -755,3 +755,79 @@ func TestImpDropOrcaWorkspaceIfRequestedNoop(t *testing.T) {
 		t.Fatalf("plain repo must not drop anything: %+v", log.Entries)
 	}
 }
+
+// Issue #248 required change 3: a wall kill on an attempt that streamed
+// real work must NOT be classified "Transient infra failure" — it gets
+// its own wall-timeout (productive) row, still retries with the same
+// backoff, and does not consume the logic-attempt budget.
+func TestImpRetryLoopProductiveWallKillIsNotInfraTransient(t *testing.T) {
+	repo := t.TempDir()
+	log := &impCaptureLog{}
+	spawns := 0
+	env := impBaseRetryEnv(repo)
+	env.spawn = func(workers.WorkerSpawnOptions) workers.WorkerResult {
+		spawns++
+		if spawns == 1 {
+			// The loop-176 attempt-1 signature: killed at the wall, but
+			// the stream classifier saw real work before the kill.
+			return workers.WorkerResult{
+				ExitCode:        124,
+				TimedOut:        true,
+				DurationMs:      4_316_000,
+				ClockResets:     112,
+				MeaningfulBytes: 631_882,
+			}
+		}
+		return workers.WorkerResult{ExitCode: 0, ResultText: "recovered"}
+	}
+	env.testGate = func(string, int) (gates.GateResult, error) {
+		return gates.GateResult{Passed: true}, nil
+	}
+	env.backoff = func(int) int { return 0 }
+	res, succeeded, err := impRunRetryLoop(env, impTestPlan(), StageConfig{TimeoutMs: 1000}, log)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !succeeded || !res.OK || res.Attempts != 1 {
+		t.Fatalf("productive wall-kill retries must not consume the logic budget: %+v", res)
+	}
+	if spawns != 2 {
+		t.Fatalf("expected 2 spawns, got %d", spawns)
+	}
+	if !log.has("warn", "wall-timeout (productive)") {
+		t.Fatalf("missing productive wall-kill row: %+v", log.Entries)
+	}
+	if log.has("warn", "Transient infra failure") {
+		t.Fatalf("productive wall-kill must not be classified infra-transient: %+v", log.Entries)
+	}
+	// The attempt-finished row must carry the kill evidence.
+	if v, ok := log.kvOf("info", "Attempt 1 finished", "clockResets"); !ok || v != 112 {
+		t.Fatalf("missing clockResets KV on attempt row: %v %v", v, ok)
+	}
+	// No transient class may be recorded for a productive wall-kill.
+	if _, err := os.Stat(impProxyStatePath(repo)); !os.IsNotExist(err) {
+		t.Fatalf("productive wall-kill must not write proxy-state.json: %v", err)
+	}
+}
+
+// A zero-progress wall kill (no meaningful bytes, no clock resets) stays
+// transient infra — the wedged-provider burn class must keep its
+// existing classification. Cold-start kills stay transient regardless of
+// stream bytes: nothing productive ever started.
+func TestImpIsInfraTransient_WallKillClasses(t *testing.T) {
+	productive := workers.WorkerResult{TimedOut: true, ClockResets: 3, MeaningfulBytes: 4096}
+	if impIsInfraTransient(productive) {
+		t.Fatalf("productive wall-kill must not be infra-transient")
+	}
+	burn := workers.WorkerResult{TimedOut: true}
+	if !impIsInfraTransient(burn) {
+		t.Fatalf("zero-progress wall-kill must stay infra-transient")
+	}
+	coldStart := workers.WorkerResult{TimedOut: true, ColdStart: true, ClockResets: 0, MeaningfulBytes: 99999}
+	if !impIsInfraTransient(coldStart) {
+		t.Fatalf("cold-start kill must stay infra-transient even with bytes")
+	}
+	if impIsProductiveWallKill(workers.WorkerResult{TimedOut: false, ClockResets: 9}) {
+		t.Fatalf("a non-timed-out result is never a wall kill")
+	}
+}

internal/workers/adapter_common.go

@@ -99,6 +99,19 @@ func derefSpawn(r *SpawnCliResult) SpawnCliResult {
 	return SpawnCliResult{}
 }
 
+// stampStreamMetrics copies the launch's Q34/Q33 stream evidence onto the
+// finalized WorkerResult (issue #248 required change 3). The metrics of
+// the FINAL attempt are the classification signal: a wall kill on an
+// attempt that streamed work is a productive wall-kill; a no-progress
+// watchdog firing on the last attempt is a burn regardless of earlier
+// attempts.
+func stampStreamMetrics(res WorkerResult, run SpawnCliResult) WorkerResult {
+	res.WatchdogFired = run.WatchdogFired
+	res.ClockResets = run.ClockResets
+	res.MeaningfulBytes = run.MeaningfulBytes
+	return res
+}
+
 // parseJSONAny parses s into any (TS JSON.parse); err swallowed by caller.
 func parseJSONAny(s string) (v any, ok bool) {
 	if err := json.Unmarshal([]byte(s), &v); err != nil {

internal/workers/claudecode.go

@@ -77,7 +77,7 @@ func (a *ClaudeCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult {
 		}
 		spawnOpts := SpawnCliOptions{
 			Dir:                 opts.Cwd,
-			TimeoutMs:           opts.TimeoutMs,
+			TimeoutMs:           launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)),
 			Env:                 opts.Env,
 			NoProgressTimeoutMs: intPtrIf(noProgressTimeoutMs != 0, noProgressTimeoutMs),
 			ColdStartTimeoutMs:  opts.ColdStartTimeoutMs,
@@ -153,7 +153,7 @@ func (a *ClaudeCodeAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult {
 		}
 	}
 
-	return claudeFinalize(derefSpawn(last), a.nowMs()-start)
+	return stampStreamMetrics(claudeFinalize(derefSpawn(last), a.nowMs()-start), derefSpawn(last))
 }
 
 // claudeBaseArgs mirrors the TS baseArgs. claude-code ignores variant;

internal/workers/grok.go

@@ -528,7 +528,7 @@ func (a *GrokAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult {
 		}
 		prepared, err := a.prepare("grok", args, SpawnCliOptions{
 			Dir:                 opts.Cwd,
-			TimeoutMs:           opts.TimeoutMs,
+			TimeoutMs:           launchBudgetMs(wallDeadline, a.nowMs(), int64(opts.TimeoutMs)),
 			Env:                 spawnEnv,
 			NoProgressTimeoutMs: &noProgressTimeoutMs,
 			WatchdogLedger:      opts.WatchdogLedger,
@@ -575,7 +575,7 @@ func (a *GrokAdapter) Spawn(opts WorkerSpawnOptions) WorkerResult {
 		fallback := grokFallbackEmpty()
 		last = &fallback
 	}
-	result := grokFinalize(*last, sessionId, a.nowMs()-start)
+	result := stampStreamMetrics(grokFinalize(*last, sessionId, a.nowMs()-start), *last)
 	// FR-GROK-03: persist the exact cost onto the run ledger for every grok
 	// worker run the provider priced. Best-effort — a ledger write must
 	// never fail the run. Identity comes from the dispatcher's watchdog