mirror of
https://github.com/Twigpine/zero.git
synced 2026-11-11 23:40:28 -10:00
KRATOS
1c30f4c117
* 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.
72 lines
2.0 KiB
Go
72 lines
2.0 KiB
Go
package cron
|
|
|
|
import (
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
)
|
|
|
|
// Mutate must serialize the read-modify-write across concurrent callers via the
|
|
// cross-process lock: N concurrent FireCount++ must all land (final == N). Without
|
|
// the lock, racing read-modify-writes would lose updates (final < N).
|
|
func TestMutateIsAtomicUnderConcurrency(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)
|
|
}
|
|
|
|
const goroutines = 20
|
|
var wg sync.WaitGroup
|
|
start := make(chan struct{})
|
|
for i := 0; i < goroutines; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
<-start
|
|
_, merr := store.Mutate(job.ID, func(current Job, readErr error) (Job, error) {
|
|
if readErr != nil {
|
|
return Job{}, readErr
|
|
}
|
|
current.FireCount++
|
|
return current, nil
|
|
})
|
|
if merr != nil {
|
|
t.Errorf("Mutate: %v", merr)
|
|
}
|
|
}()
|
|
}
|
|
close(start)
|
|
wg.Wait()
|
|
|
|
got, err := store.Get(job.ID)
|
|
if err != nil {
|
|
t.Fatalf("Get: %v", err)
|
|
}
|
|
if got.FireCount != goroutines {
|
|
t.Fatalf("FireCount = %d, want %d (lost updates → the lock is not serializing the read-modify-write)", got.FireCount, goroutines)
|
|
}
|
|
}
|
|
|
|
// A job removed between fire and persist must abort Mutate (no resurrection).
|
|
func TestMutateRemovedJobReturnsNotFound(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)
|
|
}
|
|
_, err = store.Mutate(job.ID, func(current Job, readErr error) (Job, error) {
|
|
return current, nil
|
|
})
|
|
if !errors.Is(err, ErrJobNotFound) {
|
|
t.Fatalf("Mutate on removed job = %v, want ErrJobNotFound", err)
|
|
}
|
|
// And the lock file left no litter / did not resurrect the job dir.
|
|
if _, gerr := store.Get(job.ID); !errors.Is(gerr, ErrJobNotFound) {
|
|
t.Fatalf("job should remain removed, Get = %v", gerr)
|
|
}
|
|
}
|