bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: align writer contract with schema-2

This commit is contained in:
harshitsinghbhandari 2026-05-29 16:13:14 +05:30
parent fb7dbbbb42
commit 1376b01933
8 changed files with 100 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
backend/internal/domain/lifecycle.go
View File

@ -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"`

14
backend/internal/lifecycle/fakes_test.go
View File

@ -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
}

54
backend/internal/lifecycle/manager.go
View File

@ -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 {

5
backend/internal/lifecycle/manager_test.go
View File

@ -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 ----

26
backend/internal/ports/outbound.go
View File

@ -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

12
backend/internal/session/fakes_test.go
View File

@ -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 {

3
backend/internal/session/manager.go
View File

@ -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)
}

12
backend/internal/session/manager_test.go
View File

@ -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 {