Files
zero/internal/cron/append_run_test.go
KRATOS 1c30f4c117 audit fixes: cron lock, accounting race, O(n²) hot paths, /style wiring (#260)
* audit fixes: cron lock, accounting race, O(n^2) hot paths, /style wiring

Fixes five verified findings from a code audit. Two related findings were already
addressed upstream and three need larger/separate work (noted below).

H-1 cron store: cross-process lock for the read-modify-write.
Two schedulers seeing a job due raced the metadata read-modify-write; fireJob
even documented it ("full atomicity needs file locking"). Adds a per-job O_EXCL
lock (mirroring internal/oauth's pattern, stale-reclaim included) and a Mutate
primitive doing the re-read+modify+write atomically. Update/Remove lock too. The
lock is held only for the millisecond RMW, never across a job's exec.

M-5 specialist accounting: close the check-then-append race.
specialistEventExists (unlocked ReadEvents) + a later AppendEvent let two
finishers (TaskOutput poll vs onExit, even cross-process) both pass the check and
write duplicate stop/usage events. Adds sessions.Store.AppendEventUnlessExists
(check + append atomic under the session lock) and routes the stop and usage
paths through it.

M-2 hooks audit: O(1) sequence via tail read.
append re-read and scanned the whole audit log every call to find the max
sequence (O(n^2) over a session). Reads only the file tail now; multi-process
safe (the log is a shared global file), unlike an in-memory counter which would
collide sequences across processes.

M-4 transcript rehydration: O(n^2) -> O(n).
The three resume/rewind/compact rehydration loops called appendTranscriptRow per
event, each re-scanning every existing row for the keyed-row dedup. A bulk
appendTranscriptRowsDedup builds the key set once; semantics are identical.

M-6 /style: actually shape responses.
The TUI /style command stored a style but never threaded it anywhere. Adds
agent.Options.ResponseStyle, a system-prompt directive for concise/explanatory/
review, and wires m.responseStyle through the run. "balanced" is a no-op (prompt
byte-identical to before).

Already addressed upstream (no change):
- M-1 forked-session double-count: Store.Fork already skips EventUsage events,
  with a comment naming the exact double-count it prevents.
- M-3 lineage O(N^2): Tree already does a single store.List() scan and indexes
  children in memory; no path calls List per node.

Deferred (need larger/separate changes):
- M-7 escalate_model in the TUI needs a ModelSwitcher that builds a provider
  mid-run; registering the tool alone yields a tool that errors on call.
- L-1 MCP SSE reconnect is a stream-lifecycle change (low-likelihood after the
  8 MiB per-event cap).
- L-2 dead-code removal belongs in a dedicated cleanup sweep.

Tests: per-fix unit tests including concurrency (cron Mutate, sessions
AppendEventUnlessExists) under -race; full go test ./... green.

* fix(windows): treat ERROR_ACCESS_DENIED as lock contention (cron + oauth)

The Windows smoke run failed TestMutateIsAtomicUnderConcurrency with
"...lock: Access is denied" (FireCount 19/20). On Windows a concurrent holder's
os.Remove leaves the lock file in a delete-pending state, so an O_EXCL create
races it with ERROR_ACCESS_DENIED (os.ErrPermission) rather than ErrExist — the
retry loop only treated ErrExist as contention and hard-failed.

Treat os.ErrPermission as contention (retry) in the new cron lock and in the
oauth lock it mirrors (oauth's own concurrent test exposed the same pre-existing
bug under the added load). Unix is unaffected — it returns ErrExist.

* fix: address review — cross-process atomic audit append + AppendRun guard

Addresses the review feedback on this PR.

P1 (blocking): AuditStore.append is now cross-process atomic.
The O(1) tail read observes another process's prior appends, but the read-then-
append was not atomic across processes — two could both read sequence N and write
N+1. Wraps lastSequence()+append in a per-file O_EXCL lock (hooks/lock.go, same
stale-reclaim + Windows ErrPermission handling as cron/oauth); store.mu stays as
the in-process fast path. Adds TestAuditStoreSequenceConcurrent (4 stores x 12
goroutines on one file -> unique dense sequences), which the prior sequential
test could not catch. Walked back the comment that oversold the tail read.

P2: AppendRun no longer resurrects a removed job.
fireJob's AppendRun ran after Mutate released the per-job lock; a Remove in that
window let AppendRun's MkdirAll recreate the dir with an orphaned runs.jsonl.
AppendRun now takes the per-job lock and bails if metadata.json is gone. Adds
TestAppendRunDoesNotResurrectRemovedJob.

P2: lastSequence ReadAt tolerates a short final read.
ReadAt may return io.EOF with n < len(buf) if the file shrank between Stat and
the read; treat EOF as non-fatal and use buf[:n] (matches lastEventSequence).

P3: oauth stale-reclaim no-op removed.
The double os.ReadFile "stale == data" guard compared the file to itself read
twice in a row (always true); staleness is decided by the mtime check, so the
guard is dropped (behavior-preserving).

Tests under -race; cron/hooks/oauth green. A shared filelock helper would DRY the
three near-identical locks — left as a follow-up to avoid churning the
already-green cron/oauth packages here.
2026-06-19 21:11:34 +05:30

35 lines
1013 B
Go

package cron
import (
"os"
"testing"
)
// AppendRun on a job removed mid-run must NOT resurrect its directory (which the
// old unconditional MkdirAll did, leaving an orphaned runs.jsonl with no
// metadata.json).
func TestAppendRunDoesNotResurrectRemovedJob(t *testing.T) {
store := NewStore(StoreOptions{RootDir: t.TempDir()})
job, err := store.Add(Job{Expr: "* * * * *", Prompt: "x"})
if err != nil {
t.Fatalf("Add: %v", err)
}
if err := store.Remove(job.ID); err != nil {
t.Fatalf("Remove: %v", err)
}
if err := store.AppendRun(job.ID, RunRecord{JobID: job.ID, At: store.now()}); err != nil {
t.Fatalf("AppendRun after Remove should be a no-op, got: %v", err)
}
if _, err := os.Stat(store.jobDir(job.ID)); !os.IsNotExist(err) {
t.Fatalf("AppendRun resurrected the removed job directory (stat err = %v)", err)
}
runs, err := store.Runs(job.ID)
if err != nil {
t.Fatalf("Runs: %v", err)
}
if len(runs) != 0 {
t.Fatalf("expected no runs for a removed job, got %d", len(runs))
}
}