From 38f0689ae35c4045b8b251fa0bc3d4ba576cc4f2 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 21:49:35 -0700 Subject: [PATCH] P6 task 03: an explicit Design section (state, worker loop, Cancel, Close) after the first attempt spent its whole reply deliberating the concurrency Co-Authored-By: Claude Opus 5.5 --- docs/plans/p6/03-runner-workers.md | 81 ++++++++++++++++++++++++++++++ docs/plans/p6/README.md | 6 +++ 2 files changed, 87 insertions(+) diff --git a/docs/plans/p6/03-runner-workers.md b/docs/plans/p6/03-runner-workers.md index 6862e19..b285e74 100644 --- a/docs/plans/p6/03-runner-workers.md +++ b/docs/plans/p6/03-runner-workers.md @@ -28,6 +28,87 @@ All the types in the task 02 file: `Options`, `Target`, `Result`, `TargetError`, its jobs and each job's caller; the dcache API it uses is listed in the task 02 file's "What already exists". +## Design (follow it; do not redesign) + +The concurrency is decided here so you can write it straight into `worker.go`. Put the design into +code, not into your reply. + +**State**, added to `Runner` (keep task 02's fields): + +```go + mu sync.Mutex // guards everything below + cond *sync.Cond // on mu + live []*liveJob // the rotation, in submit order + cursor int // index in live where the next scan starts + closed bool + ctx context.Context // the FetchFunc context; Close cancels it + stop context.CancelFunc + wg sync.WaitGroup // the workers + inflight int +``` + +```go +type liveJob struct { + id string + caller *auth.Caller + targets []Target // as submitted + created time.Time + next int // index of the next target to fetch + running bool // a target of this job is in flight + started bool // set to "running" in the store already + cancelled bool + retryAt time.Time // zero, or when the busy target may be tried again + busySince time.Time // zero, or when the current target first came back busy + done, failed int +} +``` + +**Submit** (task 02's code, plus): after storing, lock `mu`, append a `liveJob`, `Hooks.Depth(len(live))`, +`cond.Broadcast()`, unlock. + +**A worker** (`New` starts `cfg.Workers` of them with `wg.Add`): + +``` +loop: + lock mu + j := pick(now) // scan live from cursor, wrapping once: the first job with !running, + // !cancelled, next < len(targets), and now >= retryAt; set cursor to + // the index after it + if j == nil: + if closed: unlock; return + if some live job has a future retryAt: time.AfterFunc(until the earliest, cond.Broadcast) + cond.Wait(); unlock-free continue (Wait re-locks; go back to loop start with mu held) + j.running = true; inflight++; Hooks.Inflight(inflight) + if !j.started: SetBatchJobStatus(j.id, "running", "", now); j.started = true + unlock + res, err := call(ctx, j.caller, j.targets[j.next]) // recovers a panic into "internal error" + lock mu + j.running = false; inflight--; Hooks.Inflight(inflight) + if errors.Is(err, ErrRetryLater): + if j.busySince is zero: j.busySince = now + if now − busySince <= RetryFor: j.retryAt = now + RetryWait; Broadcast; unlock; continue + // else fall through and store it failed with err.Error() + store target j.next (rule 2); j.next++; j.busySince, j.retryAt = zero; count done/failed + Hooks.Target(...); EvictBatchOldest(MaxBytes) + if j.cancelled or j.next == len(targets): + if !j.cancelled: set the terminal status (rule 3) + remove j from live (fix cursor), forget its caller, Hooks.Depth(len(live)) + Broadcast; unlock +``` + +Store calls happen under `mu`. They are short (one SQLite connection), and holding `mu` keeps +the store consistent with the in-memory state. + +**Cancel:** lock `mu`. If the job is in `live`: set `j.cancelled`, `SetBatchJobStatus(cancelled)`, +and `Hooks.Job("cancelled")`. If `!j.running`, remove it from `live` now; a running one is +removed by its worker after the store (above), which must not overwrite `cancelled`. Broadcast, +unlock. A job not in `live` keeps task 02's behaviour. + +**Close:** lock; `closed = true`; `stop()`; Broadcast; unlock; `wg.Wait()`. Safe to call twice. + +A target stays `pending` in the store while it is in flight. The store has no `running` target +status write, and no test needs one. + ## Rules 1. **Workers.** `Workers` goroutines share one rotation of live jobs in submit order. A worker diff --git a/docs/plans/p6/README.md b/docs/plans/p6/README.md index 55a8bcb..2e92e41 100644 --- a/docs/plans/p6/README.md +++ b/docs/plans/p6/README.md @@ -78,4 +78,10 @@ intended reasons before each task is implemented. - `guard.PreCheck` moved to an owner commit on master. The old task 05 (now 06) told the implementer to edit `internal/guard/`, which `AGENTS.md` forbids: my contradiction, found while fixing the above. +- **Task 03 (workers), first attempt, 21:31–21:48.** Same mode again: the model deliberated over + the concurrency design in one reply, which ended mid-sentence. That was probably the 32k + output cap, and a reply without a tool call ends `opencode run`. No file was written. + `AGENTS.md` rule 6 alone did not prevent it. The owner changes: task 03 gains a **Design** + section that decides the concurrency (state, the worker loop, Cancel, Close), so the task is + transcription, not design. Task 02 (core) passed first time (`e2ddb2e`).