Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@internal/persis/apikey/store.go`:
- Around line 142-179: The read of existingRec/existingStored happens before
s.mu is acquired, leaving the byName index vulnerable to races; move the lock to
cover the entire read-modify-write path (i.e., acquire s.mu at the start of the
Update flow before calling s.col.Get and persis.Decode, perform the name
conflict check and call s.col.Put while holding the lock, then update s.byName
and release the lock) so that s.col.Get, the conflict check (the taken/id
comparison), the Put, and the subsequent delete/assignment to s.byName are all
serialized; alternatively, if you prefer a narrower change, keep the initial
read but re-check existingStored.Name and the byName conflict under s.mu after
locking before calling s.col.Put and before mutating s.byName to ensure no race
can produce stale byName entries.
- Around line 212-239: UpdateLastUsed performs an unsynchronized
read-modify-write against s.col (uses s.col.Get and s.col.Put) and can race with
Delete/Update; acquire the store mutex (s.mu) to serialize this path: after the
id == "" check, call s.mu.Lock() and defer s.mu.Unlock() so the Get -> Decode ->
encode -> Put sequence is atomic, ensuring you hold the lock while reading the
record and writing it back (Release only after s.col.Put completes) to prevent
clobbering or resurrecting deleted records.

In `@internal/persis/file/collection_test.go`:
- Around line 264-292: freshCollection is a package-global that gets reassigned
in TestFileCollection and TestMemoryCollection (both t.Parallel()), causing
races and flaky mixing; remove the shared global by making the factory local and
pass it into RunCollectionContract (or change RunCollectionContract to accept a
func(t *testing.T) persis.Collection), e.g. define a local freshCollection
closure inside TestFileCollection and inside TestMemoryCollection and call
RunCollectionContract(t, b.Collection("test"), freshCollection) (or update
RunCollectionContract signature to accept the factory) and update all callers to
use the passed-in factory instead of the package-level freshCollection.

In `@internal/persis/file/collection.go`:
- Around line 191-193: Sanitize and enforce a root-boundary check when mapping
IDs to disk paths: have filePath use idToRelPath(id) but then run filepath.Clean
and ensure the resulting joined path is not absolute and is contained inside
c.dir by using filepath.Rel(c.dir, joined) and rejecting if rel starts with ".."
or equals ".." (or returns an error); apply the same pattern to the other places
that join IDs/prefixes (the code around the blocks referenced at 273-287,
291-294) so no ".." or absolute paths can escape c.dir; replace any raw string
prefix checks (the check around line ~371) with a secure containment check using
filepath.Rel or by comparing path elements with filepath.Separator (e.g.,
compute rel, confirm it doesn't start with ".." and then use that for
prefix/sibling directory decisions).
- Around line 175-186: In Claim, don't swallow filesystem errors: when iterating
candidates, change the read/remove error handling so that if c.readFile(path) or
os.Remove(path) returns an error you check for os.IsNotExist and only continue
for not-found; for any other error return that error immediately instead of
continuing; after successful read and remove call
removeEmptyDirs(filepath.Dir(path), c.dir) and return fr.toRecord(), nil; only
return persis.ErrNotFound if every candidate either didn't exist (IsNotExist)
and none produced a record.

In `@internal/persis/secret/store.go`:
- Around line 209-250: The code computes oldRef/newRef before acquiring s.mu,
which allows a race where the stored record changes between Get and Lock and
leaves s.byRef inconsistent; to fix, move the critical ref computation and
existence-checks inside the mutex (i.e., acquire s.mu.Lock() immediately after
reading and decoding existingRec), then recompute
oldRef/refKey(existing.Secret.Workspace, existing.Secret.Ref) and
newRef/refKey(updated.Workspace, updated.Ref) while holding the lock, perform
the taken-check against s.byRef and apply delete/insert to s.byRef and update
existing.Secret only under the same lock so byRef remains synchronized with the
stored record for functions/methods manipulating s.byRef (e.g., Update/Delete
using sec.ID, existing, existingRec, s.col).
- Around line 285-313: WriteValue currently does a read-modify-write (s.col.Get
-> appendVersion -> s.col.Put) which causes a lost-update race under
concurrency; change it to an optimistic-transaction/retry pattern: read the
record (s.col.Get), compute the new storedRecord via appendVersion, then attempt
an atomic conditional update (compare-and-swap) against the original metadata
(e.g., UpdatedAt or record ID/ETag) instead of unconditional s.col.Put; if the
conditional Put/CompareAndSwap fails due to a concurrent update, re-read the
record and retry the appendVersion/conditional update loop a few times before
returning a conflict error. Ensure you reference the same storedRecord, rec
(Original persis.Record metadata like UpdatedAt/CreatedAt/ID), appendVersion and
s.col methods so the fix replaces the unconditional Put with an atomic
conditional update and retry logic.
- Around line 371-384: The loadByID function can return a storedRecord with a
nil Secret, leading to panics in callers like GetByID/GetCurrentVersion; update
loadByID (the function) to validate that the decoded storedRecord (sr) has a
non-nil Secret (sr.Secret) and return a clear not-found error (e.g.,
secret.ErrNotFound or fmt.Errorf("secret store: missing secret payload %q", id))
when it is nil instead of returning &sr, nil so callers that dereference
sr.Secret are protected.

In `@internal/persis/session/store.go`:
- Around line 188-227: UpdateSession currently does an unsynchronized
Get→mutate→Put causing lost updates under concurrency; wrap the entire
read-modify-write sequence in the store-level lock (or a dedicated per-session
lock) so the Get, mutation of storedSession, persis.Encode/Put and subsequent
in-memory updates (s.updatedAt and s.sortUserSessions) are atomic. Apply the
same locking pattern to AddMessage (and the other similar read-modify-write path
mentioned) so both methods acquire the lock before Get and release it only after
Put and in-memory state updates to prevent clobbered messages or stale metadata.

In `@internal/persis/testutil/memory.go`:
- Around line 168-174: copyRecord currently shallow-copies persis.Record but
leaves the ExpiresAt pointer aliased, allowing external mutations to leak;
modify copyRecord to deep-copy ExpiresAt as well: if r.ExpiresAt != nil allocate
a new value, copy the pointed value into it and assign it to cp.ExpiresAt
(analogous to how Data is duplicated) so the returned *persis.Record has its own
independent ExpiresAt.

In `@internal/persis/user/store.go`:
- Around line 189-247: The code fetches and decodes the existing record
(s.col.Get + persis.Decode and computation of oldOIDCKey) before acquiring
s.mu.Lock in the Update (and similarly in Delete), which can yield stale index
keys during concurrent writes; move the Get + persis.Decode + old key
computation (the calls to s.col.Get, persis.Decode and oidcKey for old values)
so they occur while holding s.mu.Lock (i.e., acquire s.mu.Lock before
fetching/decoding and computing oldOIDCKey) and keep the existing index-update
logic that modifies byUsername and byOIDCIdentity under the same lock; apply the
same change to the Delete code path (lines referenced in the review) to ensure
atomicity of read-modify-write on the indices.

In `@internal/persis/watermark/store.go`:
- Around line 72-76: In Store.Save, add a nil-check at the top to reject a nil
state and return a clear error instead of encoding it; specifically, in the
Store.Save(ctx context.Context, state *scheduler.SchedulerState) method validate
state != nil and return an error like "watermark store: nil state" (or similar)
before calling persis.Encode so you never persist a nil/empty payload that later
appears as corrupted in Load.

In `@internal/persis/webhook/store.go`:
- Around line 165-207: The existing code fetches and decodes the existing record
(calls to s.col.Get and persis.Decode producing existingStored) outside the
s.mu.Lock() that protects the byDAGName map, which can produce stale-byDAGName
races; move the Get + Decode (the existingRec and existingStored population) to
occur while holding s.mu.Lock() (i.e., call s.mu.Lock() before s.col.Get and
persis.Decode and keep the lock through the subsequent checks/Put and byDAGName
mutation), and apply the same change to the other similar block around the code
referenced at 217-236 so all reads/validation of existingStored and updates to
s.byDAGName happen under the same write lock.
- Around line 255-282: UpdateLastUsed performs an unlocked read-modify-write and
can be overwritten by concurrent Delete/Update calls; protect the whole
operation by serializing with the store's mutex (e.g., lock s.mu or the store's
existing sync.RWMutex) so that Get/Decode/modify/Encode/Put are executed under
the lock; acquire the write lock at the start of UpdateLastUsed and release it
after the Put (or use defer Unlock) to prevent stale overwrites or recreation
during concurrent mutations.

In `@internal/persis/workerheartbeat/store.go`:
- Around line 93-107: DeleteStale currently lists all records and then deletes
by WorkerID, opening a TOCTOU window between s.List and s.col.Delete where a
worker could refresh and be wrongly removed; change the logic to perform a
conditional/atomic delete based on the stored heartbeat timestamp or version
instead of a blind Delete. Update Store.DeleteStale to call a new or existing
conditional method on s.col (e.g.,
DeleteIfOlder/DeleteIfMatchTimestamp/DeleteIfVersion) passing r.WorkerID and
r.LastHeartbeatTime() (or version) so the backend only deletes when the stored
last-heartbeat is still <= before; if your storage backend lacks such an
operation, implement the check-and-delete inside a single transaction or loop
using Get+Delete in a single transactional API rather than separate List and
Delete calls. Ensure error handling still treats persis.ErrNotFound specially
and increments removed only when the conditional delete actually succeeded.

In `@refactor_persis_layer.html`:
- Around line 175-182: In the Mermaid HTML text nodes (e.g., the block
containing "RT --> FDR", "SC --> FQ & FP", "AG --> FSS & FA & FEV", etc.)
replace each raw arrow operator '-->' with escaped form '--&gt;' (and any other
'>' in edge operators) so the '>' is HTML-escaped; apply the same replacement to
the other Mermaid block noted (the one around the 932-937 region) so the
spec-char-escape lint rule passes.
- Around line 290-291: The rendered inline comment text for the fields Since and
Until uses raw comparison operators (">=" and "<") which trigger
spec-char-escape; update the comment strings for Since (*time.Time // CreatedAt
>= Since) and Until (*time.Time // CreatedAt < Until) to use HTML entities (e.g.
&gt;= and &lt;) so the operators are escaped in the HTML-rendered code snippet
and no parser warnings occur.

---

Nitpick comments:
In `@internal/persis/session/store_test.go`:
- Around line 115-167: Add a new concurrent test (e.g.,
TestAddMessage_Concurrent) that uses newStore and newSession to create a
session, then launches many goroutines that call s.AddMessage(ctx, "s1",
&agent.Message{SequenceID: i, Type: agent.MessageTypeUser, Content:
fmt.Sprintf("m%d", i)}) concurrently (use sync.WaitGroup), collect/require no
errors from each goroutine, wait for completion, then call s.GetMessages(ctx,
"s1") and assert the returned slice length equals the number of goroutines and
that all expected contents or sequence IDs are present; also call
s.GetLatestSequenceID(ctx, "s1") and assert it equals the max SequenceID to
ensure no messages were lost.

In `@internal/persis/watermark/store_test.go`:
- Around line 25-85: The tests only cover basic Save/Load and overwrite paths;
add unit tests that exercise Load's migration and corrupt-state recovery
branches by writing legacy and invalid payloads into the same backend store and
asserting Load's behavior. Specifically, add tests (e.g.,
TestLoad_MigrateLegacyVersions, TestLoad_UnknownVersionRecovery,
TestLoad_DecodeFailureFallback) that use newStore(t) or the store backend to
seed stored bytes representing versions 0/1/2, an unknown version value, and a
deliberately corrupted/invalid encoded blob, then call s.Load(ctx) and assert
the expected migration result or recovery behavior (version normalization to
scheduler.SchedulerStateVersion, DAG contents/LastTick for migrated cases, and
the fallback/reset behavior for unknown or decode failures) to lock in the
existing Load, Save, and migration logic.

In `@internal/persis/workerheartbeat/store_test.go`:
- Around line 79-104: Add a regression test (e.g.,
TestDeleteStale_RefreshDuringCleanup) that creates a stale record using
exec.WorkerHeartbeatRecord and s.Upsert, then simulates a concurrent refresh by
calling s.Upsert again for that same WorkerID with an updated LastHeartbeatAt
(now) before invoking s.DeleteStale with the cutoff; assert DeleteStale returns
0 deletions for that worker and that s.Get still finds the refreshed record,
using the existing DeleteStale, Upsert, and Get methods to locate behavior.