From 1376b019332f45e3d9b619fde4eff42235c5be7e Mon Sep 17 00:00:00 2001 From: harshitsinghbhandari <24b4506@iitb.ac.in> Date: Fri, 29 May 2026 16:13:14 +0530 Subject: [PATCH] fix: align writer contract with schema-2 --- backend/internal/domain/lifecycle.go | 9 ++-- backend/internal/lifecycle/fakes_test.go | 14 +++++- backend/internal/lifecycle/manager.go | 54 ++++++++++++++++------ backend/internal/lifecycle/manager_test.go | 5 +- backend/internal/ports/outbound.go | 26 +++++++++-- backend/internal/session/fakes_test.go | 12 +++-- backend/internal/session/manager.go | 3 +- backend/internal/session/manager_test.go | 12 ++--- 8 files changed, 100 insertions(+), 35 deletions(-) diff --git a/backend/internal/domain/lifecycle.go b/backend/internal/domain/lifecycle.go index ab22eed31..fca87b6b8 100644 --- a/backend/internal/domain/lifecycle.go +++ b/backend/internal/domain/lifecycle.go @@ -20,10 +20,11 @@ const LifecycleVersion = 1 // between observations (they are read back by the pure decide core), so they // live in the persisted record too. type CanonicalSessionLifecycle struct { - // Version is the schema version of this record's shape (LifecycleVersion). - Version int `json:"version"` - // Revision is a monotonic counter the LCM bumps on every full-row Upsert and - // is distinct from the schema Version above. + // Version is the Go-only schema-shape constant for this record. It is not + // persisted and is not part of the CDC payload. + Version int + // Revision is the per-write monotonic counter. The storage layer's Upsert + // bumps it when the full row is persisted; the LCM does not. Revision int `json:"revision"` Session SessionSubstate `json:"session"` PR PRSubstate `json:"pr"` diff --git a/backend/internal/lifecycle/fakes_test.go b/backend/internal/lifecycle/fakes_test.go index af7c06559..5bacb49a6 100644 --- a/backend/internal/lifecycle/fakes_test.go +++ b/backend/internal/lifecycle/fakes_test.go @@ -2,6 +2,7 @@ package lifecycle import ( "context" + "fmt" "sync" "github.com/aoagents/agent-orchestrator/backend/internal/domain" @@ -45,9 +46,20 @@ func (s *fakeStore) Load(_ context.Context, id domain.SessionID) (domain.Canonic return rec.Lifecycle, true, nil } -func (s *fakeStore) Upsert(_ context.Context, rec domain.SessionRecord) error { +func (s *fakeStore) Upsert(_ context.Context, rec domain.SessionRecord, _ ports.EventType) error { s.mu.Lock() defer s.mu.Unlock() + if existing, ok := s.records[rec.ID]; ok { + if rec.Lifecycle.Revision != existing.Lifecycle.Revision { + return fmt.Errorf("revision mismatch for %s: have %d, want %d", rec.ID, rec.Lifecycle.Revision, existing.Lifecycle.Revision) + } + rec.Lifecycle.Revision = existing.Lifecycle.Revision + 1 + } else { + if rec.Lifecycle.Revision != 0 { + return fmt.Errorf("revision mismatch for insert %s: have %d, want 0", rec.ID, rec.Lifecycle.Revision) + } + rec.Lifecycle.Revision = 1 + } if rec.Lifecycle.Version == 0 { rec.Lifecycle.Version = domain.LifecycleVersion } diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index 7494e346f..bedfb3b1f 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -1,8 +1,9 @@ // Package lifecycle implements ports.LifecycleManager: the synchronous // observe->decide->persist reducer. Every Apply*/On* entrypoint runs the same // pipeline under a per-session lock — load the full canonical record, run the -// matching pure decider, diff into a full-row Upsert, persist. The LCM never -// polls and never writes the display status (that is derived on read). +// matching pure decider, classify the resulting change, and persist the full +// row through the store. The store owns Revision++; the LCM never polls and +// never writes the display status (that is derived on read). // // After a transition is persisted, the Apply* paths fire the mapped reaction // (the ACT layer: reaction table + escalation engine) via the react() chokepoint @@ -120,8 +121,8 @@ type transition struct { // mutate runs the shared pipeline: load full row -> build next canonical -> // Upsert full row (only if changed). decideFn returns the full next lifecycle -// and whether it changed anything; false is a clean no-op (no write, no revision -// bump), which is how failed-probe / unknown-fact inputs are dropped. +// and whether it changed anything; false is a clean no-op (no write), which is +// how failed-probe / unknown-fact inputs are dropped. // // On a write it returns the transition (before/after canonical) so the caller — // which still holds the originating facts — can fire the mapped reaction. @@ -144,9 +145,9 @@ func (m *Manager) mutate( if !changed { return nil } - rec.Lifecycle = m.prepareLifecycleWrite(cur, next) + rec.Lifecycle = m.prepareLifecycleWrite(next) rec.UpdatedAt = m.clock() - if err := m.store.Upsert(ctx, rec); err != nil { + if err := m.store.Upsert(ctx, rec, classifyEventType(cur, rec.Lifecycle, false)); err != nil { return err } tr = &transition{beforeLC: cur, afterLC: rec.Lifecycle} @@ -155,9 +156,8 @@ func (m *Manager) mutate( return tr, err } -func (m *Manager) prepareLifecycleWrite(cur, next domain.CanonicalSessionLifecycle) domain.CanonicalSessionLifecycle { +func (m *Manager) prepareLifecycleWrite(next domain.CanonicalSessionLifecycle) domain.CanonicalSessionLifecycle { next.Version = domain.LifecycleVersion - next.Revision = cur.Revision + 1 return next } @@ -285,10 +285,13 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID, // OnSpawnInitiated seeds or reopens the full session record for a spawn-like // mutation. It is the Session Manager's create/reopen command under the Writer // contract: the SM builds the identity + initial canonical row, but only the LCM -// writes it. +// writes it. Fresh rows emit session_created; reopening a terminal row reuses +// the current row as the before-image and lets the classifier emit the schema +// event for the reopen. func (m *Manager) OnSpawnInitiated(ctx context.Context, rec domain.SessionRecord) error { return m.withLock(rec.ID, func() error { cur := rec.Lifecycle + isInsert := true if current, ok, err := m.store.Get(ctx, rec.ID); err != nil { return err } else if ok { @@ -297,14 +300,20 @@ func (m *Manager) OnSpawnInitiated(ctx context.Context, rec domain.SessionRecord return fmt.Errorf("lifecycle: OnSpawnInitiated for active session %q", rec.ID) } cur = currentLC + isInsert = false + } + rec.Lifecycle = m.prepareLifecycleWrite(rec.Lifecycle) + if isInsert { + rec.Lifecycle.Revision = 0 + } else { + rec.Lifecycle.Revision = cur.Revision } - rec.Lifecycle = m.prepareLifecycleWrite(cur, rec.Lifecycle) now := m.clock() if rec.CreatedAt.IsZero() { rec.CreatedAt = now } rec.UpdatedAt = now - return m.store.Upsert(ctx, rec) + return m.store.Upsert(ctx, rec, classifyEventType(cur, rec.Lifecycle, isInsert)) }) } @@ -329,9 +338,9 @@ func (m *Manager) OnSpawnCompleted(ctx context.Context, id domain.SessionID, o p cur := rec.Lifecycle next := cur next.Runtime = rt - rec.Lifecycle = m.prepareLifecycleWrite(cur, next) + rec.Lifecycle = m.prepareLifecycleWrite(next) rec.UpdatedAt = m.clock() - if err := m.store.Upsert(ctx, rec); err != nil { + if err := m.store.Upsert(ctx, rec, classifyEventType(cur, rec.Lifecycle, false)); err != nil { return err } } @@ -433,6 +442,25 @@ func setDetecting(next *domain.CanonicalSessionLifecycle, d *domain.DetectingSta return false } +func classifyEventType(before, after domain.CanonicalSessionLifecycle, isInsert bool) ports.EventType { + switch { + case isInsert: + return ports.EventSessionCreated + case before.Session.State != after.Session.State && after.Session.State == domain.SessionTerminated: + return ports.EventSessionTerminated + case before.Session != after.Session: + return ports.EventSessionStateChanged + case before.PR != after.PR: + return ports.EventSessionPRUpdated + case before.Runtime != after.Runtime: + return ports.EventSessionRuntimeUpdated + case before.Activity != after.Activity: + return ports.EventSessionActivityUpdated + default: + return ports.EventSessionUpdated + } +} + // sameActivity compares activity sub-states with time-aware equality (== on // time.Time is monotonic-clock sensitive and would spuriously report changes). func sameActivity(a, b domain.ActivitySubstate) bool { diff --git a/backend/internal/lifecycle/manager_test.go b/backend/internal/lifecycle/manager_test.go index fb7bf84d5..e93a33c5d 100644 --- a/backend/internal/lifecycle/manager_test.go +++ b/backend/internal/lifecycle/manager_test.go @@ -483,7 +483,7 @@ func TestFakeStoreUpsertFullRow(t *testing.T) { } rec.Lifecycle.Session = domain.SessionSubstate{State: domain.SessionIdle, Reason: domain.ReasonResearchComplete} rec.Lifecycle.Runtime = domain.RuntimeSubstate{State: domain.RuntimeExited} - if err := store.Upsert(context.Background(), rec); err != nil { + if err := store.Upsert(context.Background(), rec, ports.EventSessionStateChanged); err != nil { t.Fatalf("upsert: %v", err) } @@ -491,6 +491,9 @@ func TestFakeStoreUpsertFullRow(t *testing.T) { if got.Lifecycle.Session.State != domain.SessionIdle || got.Lifecycle.Runtime.State != domain.RuntimeExited { t.Fatalf("upsert should replace the full canonical row, got %+v", got.Lifecycle) } + if got.Lifecycle.Revision != 1 { + t.Fatalf("upsert should bump revision inside the store, got %d want 1", got.Lifecycle.Revision) + } } // ---- per-session serialisation under the race detector ---- diff --git a/backend/internal/ports/outbound.go b/backend/internal/ports/outbound.go index c5ee6bfac..ba08d9b9a 100644 --- a/backend/internal/ports/outbound.go +++ b/backend/internal/ports/outbound.go @@ -11,15 +11,16 @@ import ( // Writer contract: the Lifecycle Manager (LCM) is the sole logical writer of // sessions. Controllers, the Session Manager, observers, and other goroutines // must route mutations to the LCM; no other goroutine writes sessions directly. -// The LCM serializes mutations and calls Upsert with the full SessionRecord. -// This full-row insert-or-update replaces the older sparse merge-patch model and -// is safe only under the single-writer LCM invariant. +// The LCM serializes mutations and calls Upsert with the full SessionRecord and +// the classified event_type. The storage layer owns Revision++ and performs the +// full-row insert-or-update; the older sparse merge-patch model is gone. // // List/Get return persistence records (no derived status); the Session Manager // hydrates them into domain.Session by attaching DeriveLegacyStatus on read. type LifecycleStore interface { - // Upsert inserts or replaces the full session row. Only the LCM may call it. - Upsert(ctx context.Context, rec domain.SessionRecord) error + // Upsert inserts or replaces the full session row and bumps Revision inside + // the storage layer. Only the LCM may call it. + Upsert(ctx context.Context, rec domain.SessionRecord, eventType EventType) error Load(ctx context.Context, id domain.SessionID) (domain.CanonicalSessionLifecycle, bool, error) List(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) GetMetadata(ctx context.Context, id domain.SessionID) (map[string]string, error) @@ -31,6 +32,21 @@ type LifecycleStore interface { Get(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) } +// EventType is the schema-level event label attached to each Upsert. +type EventType string + +const ( + EventSessionCreated EventType = "session_created" + EventSessionTerminated EventType = "session_terminated" + EventSessionStateChanged EventType = "session_state_changed" + EventSessionPRUpdated EventType = "session_pr_updated" + EventSessionRuntimeUpdated EventType = "session_runtime_updated" + EventSessionAttentionUpdated EventType = "session_attention_updated" + EventSessionActivityUpdated EventType = "session_activity_updated" + EventSessionDisplayUpdated EventType = "session_display_updated" + EventSessionUpdated EventType = "session_updated" +) + // Notifier delivers events to the human (desktop/Slack later). Push, never pull. type Notifier interface { Notify(ctx context.Context, event OrchestratorEvent) error diff --git a/backend/internal/session/fakes_test.go b/backend/internal/session/fakes_test.go index ebeac6598..1796e509e 100644 --- a/backend/internal/session/fakes_test.go +++ b/backend/internal/session/fakes_test.go @@ -59,14 +59,18 @@ func newFakeStore() *fakeStore { } } -func (s *fakeStore) Upsert(_ context.Context, rec domain.SessionRecord) error { +func (s *fakeStore) Upsert(_ context.Context, rec domain.SessionRecord, _ ports.EventType) error { s.mu.Lock() defer s.mu.Unlock() if existing, ok := s.records[rec.ID]; ok { - if rec.Lifecycle.Revision != existing.Lifecycle.Revision+1 { - return fmt.Errorf("revision mismatch for %s: have %d, want %d", rec.ID, rec.Lifecycle.Revision, existing.Lifecycle.Revision+1) + if rec.Lifecycle.Revision != existing.Lifecycle.Revision { + return fmt.Errorf("revision mismatch for %s: have %d, want %d", rec.ID, rec.Lifecycle.Revision, existing.Lifecycle.Revision) + } + rec.Lifecycle.Revision = existing.Lifecycle.Revision + 1 + } else { + if rec.Lifecycle.Revision != 0 { + return fmt.Errorf("revision mismatch for insert %s: have %d, want 0", rec.ID, rec.Lifecycle.Revision) } - } else if rec.Lifecycle.Revision == 0 { rec.Lifecycle.Revision = 1 } if rec.Lifecycle.Version == 0 { diff --git a/backend/internal/session/manager.go b/backend/internal/session/manager.go index c7ccf3396..4d9157398 100644 --- a/backend/internal/session/manager.go +++ b/backend/internal/session/manager.go @@ -320,7 +320,8 @@ func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Sess } if err := m.lcm.OnSpawnCompleted(ctx, id, outcome); err != nil { m.rollbackRuntime(ctx, handle) - // Re-upsert the original record to undo the reopen. + // Re-upsert the original record to undo the reopen; the store will + // assign the next revision. if revertErr := m.lcm.OnSpawnInitiated(ctx, rec); revertErr != nil { return domain.Session{}, fmt.Errorf("restore %s: revert after spawn completed failure: %w (original error: %v)", id, revertErr, err) } diff --git a/backend/internal/session/manager_test.go b/backend/internal/session/manager_test.go index b54cf4b09..381fb57e0 100644 --- a/backend/internal/session/manager_test.go +++ b/backend/internal/session/manager_test.go @@ -128,7 +128,7 @@ func TestSpawn_ExistingSessionIDRejectedBeforeWork(t *testing.T) { ID: "sess-1", ProjectID: testProject, Lifecycle: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.PRNone, ""), - }); err != nil { + }, ports.EventSessionCreated); err != nil { t.Fatalf("seed existing row: %v", err) } @@ -246,7 +246,7 @@ func TestKill_IncompleteMetadata_RefusesTeardown(t *testing.T) { if err := h.store.Upsert(ctx, domain.SessionRecord{ ID: "sess-1", ProjectID: testProject, Lifecycle: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.PRNone, ""), - }); err != nil { + }, ports.EventSessionCreated); err != nil { t.Fatalf("upsert: %v", err) } @@ -270,7 +270,7 @@ func TestCleanup_IncompleteMetadata_Skipped(t *testing.T) { if err := h.store.Upsert(ctx, domain.SessionRecord{ ID: "orphan-1", ProjectID: testProject, Lifecycle: lc(domain.SessionTerminated, domain.ReasonManuallyKilled, domain.PRNone, ""), - }); err != nil { + }, ports.EventSessionCreated); err != nil { t.Fatalf("upsert: %v", err) } @@ -333,7 +333,7 @@ func TestListAndGet_DeriveStatus(t *testing.T) { h := newHarness("unused") ctx := context.Background() for _, c := range cases { - if err := h.store.Upsert(ctx, domain.SessionRecord{ID: domain.SessionID(c.name), ProjectID: testProject, Lifecycle: c.lc}); err != nil { + if err := h.store.Upsert(ctx, domain.SessionRecord{ID: domain.SessionID(c.name), ProjectID: testProject, Lifecycle: c.lc}, ports.EventSessionCreated); err != nil { t.Fatalf("upsert %s: %v", c.name, err) } } @@ -517,7 +517,7 @@ func TestCleanup_SkipsUncommittedWork(t *testing.T) { if err := h.store.Upsert(ctx, domain.SessionRecord{ ID: "live-1", ProjectID: testProject, Lifecycle: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.PRNone, ""), - }); err != nil { + }, ports.EventSessionCreated); err != nil { t.Fatalf("upsert live: %v", err) } // dirty-1's worktree still holds uncommitted work — Destroy refuses it. @@ -557,7 +557,7 @@ func seedTerminal(t *testing.T, h *harness, id domain.SessionID, wsPath string) if err := h.store.Upsert(ctx, domain.SessionRecord{ ID: id, ProjectID: testProject, Lifecycle: lc(domain.SessionTerminated, domain.ReasonManuallyKilled, domain.PRNone, ""), - }); err != nil { + }, ports.EventSessionCreated); err != nil { t.Fatalf("upsert %s: %v", id, err) } if err := h.store.PatchMetadata(ctx, id, map[string]string{lifecycle.MetaWorkspacePath: wsPath}); err != nil {