diff --git a/backend/cdc_wiring.go b/backend/cdc_wiring.go index 89997e7dc..cfae4fdb9 100644 --- a/backend/cdc_wiring.go +++ b/backend/cdc_wiring.go @@ -125,7 +125,6 @@ func (s snapshotSource) Snapshot(ctx context.Context) ([]cdc.Event, int64, error events := make([]cdc.Event, 0, len(recs)) for _, r := range recs { r.Lifecycle.Version = domain.LifecycleVersion - r.Metadata = nil blob, err := json.Marshal(r) if err != nil { return nil, 0, fmt.Errorf("marshal snapshot %s: %w", r.ID, err) diff --git a/backend/internal/domain/session.go b/backend/internal/domain/session.go index 578cca406..c9cd8d963 100644 --- a/backend/internal/domain/session.go +++ b/backend/internal/domain/session.go @@ -17,16 +17,41 @@ const ( KindOrchestrator SessionKind = "orchestrator" ) +// SessionMetadata is the typed, off-canonical metadata for a session: the +// operational handles and seed inputs the Session Manager and reaper need but +// that are NOT part of the canonical lifecycle. The set of fields is fixed here +// (no free-form keys), so what a session can carry is a compile-time fact, and +// it is persisted 1:1 in the session_metadata table off the CDC path. +// +// Empty fields mean "unset": PatchMetadata never overwrites a stored value with +// an empty one, so a partial write (spawn setting only the runtime handle) does +// not clobber a value set earlier (the branch set at creation). +type SessionMetadata struct { + Branch string `json:"branch,omitempty"` + WorkspacePath string `json:"workspacePath,omitempty"` + RuntimeHandleID string `json:"runtimeHandleId,omitempty"` + RuntimeName string `json:"runtimeName,omitempty"` + AgentSessionID string `json:"agentSessionId,omitempty"` + Prompt string `json:"prompt,omitempty"` +} + +// IsZero reports whether no metadata field is set. +func (m SessionMetadata) IsZero() bool { return m == SessionMetadata{} } + // SessionRecord is the PERSISTENCE shape: identity, canonical lifecycle, and // metadata — everything the store holds, and nothing derived. The store reads // and writes records; it never produces the derived display status. +// +// Metadata is json:"-" on purpose: it lives off the canonical path, so it must +// never ride along in the change_log / snapshot payloads. Enforcing that at the +// type level means no caller has to remember to scrub it before marshalling. type SessionRecord struct { ID SessionID `json:"id"` ProjectID ProjectID `json:"projectId"` IssueID IssueID `json:"issueId,omitempty"` Kind SessionKind `json:"kind"` Lifecycle CanonicalSessionLifecycle `json:"lifecycle"` - Metadata map[string]string `json:"metadata,omitempty"` + Metadata SessionMetadata `json:"-"` CreatedAt time.Time `json:"createdAt"` UpdatedAt time.Time `json:"updatedAt"` } diff --git a/backend/internal/lifecycle/fakes_test.go b/backend/internal/lifecycle/fakes_test.go index 5bacb49a6..45aec91b9 100644 --- a/backend/internal/lifecycle/fakes_test.go +++ b/backend/internal/lifecycle/fakes_test.go @@ -14,7 +14,7 @@ import ( type fakeStore struct { mu sync.Mutex records map[domain.SessionID]*domain.SessionRecord - metadata map[domain.SessionID]map[string]string + metadata map[domain.SessionID]domain.SessionMetadata } var _ ports.LifecycleStore = (*fakeStore)(nil) @@ -22,7 +22,7 @@ var _ ports.LifecycleStore = (*fakeStore)(nil) func newFakeStore() *fakeStore { return &fakeStore{ records: map[domain.SessionID]*domain.SessionRecord{}, - metadata: map[domain.SessionID]map[string]string{}, + metadata: map[domain.SessionID]domain.SessionMetadata{}, } } @@ -90,28 +90,43 @@ func (s *fakeStore) List(_ context.Context, project domain.ProjectID) ([]domain. return out, nil } -func (s *fakeStore) GetMetadata(_ context.Context, id domain.SessionID) (map[string]string, error) { +func (s *fakeStore) GetMetadata(_ context.Context, id domain.SessionID) (domain.SessionMetadata, error) { s.mu.Lock() defer s.mu.Unlock() - out := map[string]string{} - for k, v := range s.metadata[id] { - out[k] = v - } - return out, nil + return s.metadata[id], nil } -func (s *fakeStore) PatchMetadata(_ context.Context, id domain.SessionID, kv map[string]string) error { +func (s *fakeStore) PatchMetadata(_ context.Context, id domain.SessionID, meta domain.SessionMetadata) error { s.mu.Lock() defer s.mu.Unlock() - if s.metadata[id] == nil { - s.metadata[id] = map[string]string{} - } - for k, v := range kv { - s.metadata[id][k] = v - } + s.metadata[id] = mergeSessionMetadata(s.metadata[id], meta) return nil } +// mergeSessionMetadata applies meta onto dst with the store's "empty = leave +// unchanged" semantics, so partial patches do not clobber earlier values. +func mergeSessionMetadata(dst, meta domain.SessionMetadata) domain.SessionMetadata { + if meta.Branch != "" { + dst.Branch = meta.Branch + } + if meta.WorkspacePath != "" { + dst.WorkspacePath = meta.WorkspacePath + } + if meta.RuntimeHandleID != "" { + dst.RuntimeHandleID = meta.RuntimeHandleID + } + if meta.RuntimeName != "" { + dst.RuntimeName = meta.RuntimeName + } + if meta.AgentSessionID != "" { + dst.AgentSessionID = meta.AgentSessionID + } + if meta.Prompt != "" { + dst.Prompt = meta.Prompt + } + return dst +} + // recordingNotifier captures emitted events for assertions. type recordingNotifier struct { mu sync.Mutex diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index 54e6887f0..63d7164a1 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -21,19 +21,11 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) -// Metadata keys OnSpawnCompleted records for the spawned session's handles. -// -// MetaPrompt is the assembled launch prompt, persisted so a Restore that finds -// no captured agent session id can still fall back to a fresh launch with the -// same prompt rather than failing. -const ( - MetaBranch = "branch" - MetaWorkspacePath = "workspacePath" - MetaRuntimeHandleID = "runtimeHandleId" - MetaRuntimeName = "runtimeName" - MetaAgentSessionID = "agentSessionId" - MetaPrompt = "prompt" -) +// Session metadata is now the typed domain.SessionMetadata struct (was a +// free-form string map keyed by Meta* constants). OnSpawnCompleted records the +// spawned session's handles via spawnMetadata; Prompt is the assembled launch +// prompt, persisted so a Restore that finds no captured agent session id can +// still fall back to a fresh launch with the same prompt rather than failing. // Manager is the LCM. The Apply* pipeline persists a transition and then fires // the mapped reaction via Notifier/AgentMessenger (see reactions.go). @@ -381,7 +373,7 @@ func (m *Manager) OnSpawnCompleted(ctx context.Context, id domain.SessionID, o p return err } } - if meta := spawnMetadata(o); len(meta) > 0 { + if meta := spawnMetadata(o); !meta.IsZero() { if err := m.store.PatchMetadata(ctx, id, meta); err != nil { return err } @@ -545,25 +537,13 @@ func sameActivity(a, b domain.ActivitySubstate) bool { return a.State == b.State && a.Source == b.Source && a.LastActivityAt.Equal(b.LastActivityAt) } -func spawnMetadata(o ports.SpawnOutcome) map[string]string { - meta := map[string]string{} - if o.Branch != "" { - meta[MetaBranch] = o.Branch +func spawnMetadata(o ports.SpawnOutcome) domain.SessionMetadata { + return domain.SessionMetadata{ + Branch: o.Branch, + WorkspacePath: o.WorkspacePath, + RuntimeHandleID: o.RuntimeHandle.ID, + RuntimeName: o.RuntimeHandle.RuntimeName, + AgentSessionID: o.AgentSessionID, + Prompt: o.Prompt, } - if o.WorkspacePath != "" { - meta[MetaWorkspacePath] = o.WorkspacePath - } - if o.RuntimeHandle.ID != "" { - meta[MetaRuntimeHandleID] = o.RuntimeHandle.ID - } - if o.RuntimeHandle.RuntimeName != "" { - meta[MetaRuntimeName] = o.RuntimeHandle.RuntimeName - } - if o.AgentSessionID != "" { - meta[MetaAgentSessionID] = o.AgentSessionID - } - if o.Prompt != "" { - meta[MetaPrompt] = o.Prompt - } - return meta } diff --git a/backend/internal/lifecycle/manager_test.go b/backend/internal/lifecycle/manager_test.go index 6a2cc1d1a..96557e8f0 100644 --- a/backend/internal/lifecycle/manager_test.go +++ b/backend/internal/lifecycle/manager_test.go @@ -388,7 +388,7 @@ func TestOnSpawnCompleted(t *testing.T) { t.Errorf("display = %v, want spawning", got) } meta, _ := store.GetMetadata(context.Background(), sid) - if meta[MetaBranch] != "feat/x" || meta[MetaAgentSessionID] != "agent-1" || meta[MetaRuntimeName] != "tmux" { + if meta.Branch != "feat/x" || meta.AgentSessionID != "agent-1" || meta.RuntimeName != "tmux" { t.Errorf("metadata not recorded: %+v", meta) } } diff --git a/backend/internal/observe/reaper/reaper.go b/backend/internal/observe/reaper/reaper.go index 66456ea6e..579f1d634 100644 --- a/backend/internal/observe/reaper/reaper.go +++ b/backend/internal/observe/reaper/reaper.go @@ -16,7 +16,6 @@ import ( "time" "github.com/aoagents/agent-orchestrator/backend/internal/domain" - "github.com/aoagents/agent-orchestrator/backend/internal/lifecycle" "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) @@ -203,11 +202,11 @@ func (r *Reaper) probeOne(ctx context.Context, sess domain.SessionRecord, now ti } // handleFromRecord reconstructs the RuntimeHandle stored on the session by -// OnSpawnCompleted. Both keys are required; either being empty is the +// OnSpawnCompleted. Both fields are required; either being empty is the // "session lacks a probable handle" signal that probeOne uses to skip. func handleFromRecord(rec domain.SessionRecord) (ports.RuntimeHandle, bool) { - id := rec.Metadata[lifecycle.MetaRuntimeHandleID] - name := rec.Metadata[lifecycle.MetaRuntimeName] + id := rec.Metadata.RuntimeHandleID + name := rec.Metadata.RuntimeName if id == "" || name == "" { return ports.RuntimeHandle{}, false } diff --git a/backend/internal/observe/reaper/reaper_test.go b/backend/internal/observe/reaper/reaper_test.go index d6b88efdf..0d3b4d479 100644 --- a/backend/internal/observe/reaper/reaper_test.go +++ b/backend/internal/observe/reaper/reaper_test.go @@ -9,7 +9,6 @@ import ( "time" "github.com/aoagents/agent-orchestrator/backend/internal/domain" - "github.com/aoagents/agent-orchestrator/backend/internal/lifecycle" "github.com/aoagents/agent-orchestrator/backend/internal/observe/reaper" "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) @@ -124,9 +123,9 @@ func aliveSessionWith(id domain.SessionID, runtimeName, handleID string) domain. Session: domain.SessionSubstate{State: domain.SessionWorking, Reason: domain.ReasonTaskInProgress}, Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning}, }, - Metadata: map[string]string{ - lifecycle.MetaRuntimeHandleID: handleID, - lifecycle.MetaRuntimeName: runtimeName, + Metadata: domain.SessionMetadata{ + RuntimeHandleID: handleID, + RuntimeName: runtimeName, }, } } @@ -141,9 +140,9 @@ func detectingSessionWith(id domain.SessionID, runtimeName, handleID string) dom Session: domain.SessionSubstate{State: domain.SessionDetecting, Reason: domain.ReasonProbeFailure}, Runtime: domain.RuntimeSubstate{State: domain.RuntimeProbeFailed, Reason: domain.RuntimeReasonProbeError}, }, - Metadata: map[string]string{ - lifecycle.MetaRuntimeHandleID: handleID, - lifecycle.MetaRuntimeName: runtimeName, + Metadata: domain.SessionMetadata{ + RuntimeHandleID: handleID, + RuntimeName: runtimeName, }, } } @@ -367,7 +366,7 @@ func TestReaper_SkipsMissingHandle(t *testing.T) { now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) clock := func() time.Time { return now } sess := aliveSessionWith("s1", "tmux", "h1") - delete(sess.Metadata, lifecycle.MetaRuntimeHandleID) + sess.Metadata.RuntimeHandleID = "" lcm := &fakeLCM{sessions: []domain.SessionRecord{sess}} rt := &fakeRuntime{results: map[string]aliveResult{}} rp := reaper.New(lcm, reaper.MapRegistry{"tmux": rt}, reaper.Config{Clock: clock, Tick: time.Hour}) diff --git a/backend/internal/ports/outbound.go b/backend/internal/ports/outbound.go index ba08d9b9a..c64a1e6d4 100644 --- a/backend/internal/ports/outbound.go +++ b/backend/internal/ports/outbound.go @@ -23,8 +23,8 @@ type LifecycleStore interface { 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) - PatchMetadata(ctx context.Context, id domain.SessionID, kv map[string]string) error + GetMetadata(ctx context.Context, id domain.SessionID) (domain.SessionMetadata, error) + PatchMetadata(ctx context.Context, id domain.SessionID, meta domain.SessionMetadata) error // Get returns a single full record (with identity) by id. Load is // lifecycle-only, so readers use this to build the read-model and reconstruct diff --git a/backend/internal/session/fakes_test.go b/backend/internal/session/fakes_test.go index 71eaa4afd..033f6de75 100644 --- a/backend/internal/session/fakes_test.go +++ b/backend/internal/session/fakes_test.go @@ -47,7 +47,7 @@ func (c *callLog) indexOf(name string) int { type fakeStore struct { mu sync.Mutex records map[domain.SessionID]*domain.SessionRecord - metadata map[domain.SessionID]map[string]string + metadata map[domain.SessionID]domain.SessionMetadata } var _ ports.LifecycleStore = (*fakeStore)(nil) @@ -55,7 +55,7 @@ var _ ports.LifecycleStore = (*fakeStore)(nil) func newFakeStore() *fakeStore { return &fakeStore{ records: map[domain.SessionID]*domain.SessionRecord{}, - metadata: map[domain.SessionID]map[string]string{}, + metadata: map[domain.SessionID]domain.SessionMetadata{}, } } @@ -113,30 +113,47 @@ func (s *fakeStore) List(_ context.Context, project domain.ProjectID) ([]domain. return out, nil } -func (s *fakeStore) GetMetadata(_ context.Context, id domain.SessionID) (map[string]string, error) { +func (s *fakeStore) GetMetadata(_ context.Context, id domain.SessionID) (domain.SessionMetadata, error) { s.mu.Lock() defer s.mu.Unlock() - return cloneMap(s.metadata[id]), nil + return s.metadata[id], nil } -func (s *fakeStore) PatchMetadata(_ context.Context, id domain.SessionID, kv map[string]string) error { +func (s *fakeStore) PatchMetadata(_ context.Context, id domain.SessionID, meta domain.SessionMetadata) error { s.mu.Lock() defer s.mu.Unlock() - if s.metadata[id] == nil { - s.metadata[id] = map[string]string{} - } - for k, v := range kv { - s.metadata[id][k] = v - } + s.metadata[id] = mergeSessionMetadata(s.metadata[id], meta) return nil } +// mergeSessionMetadata applies meta onto dst with the store's "empty = leave +// unchanged" semantics, so partial patches do not clobber earlier values. +func mergeSessionMetadata(dst, meta domain.SessionMetadata) domain.SessionMetadata { + if meta.Branch != "" { + dst.Branch = meta.Branch + } + if meta.WorkspacePath != "" { + dst.WorkspacePath = meta.WorkspacePath + } + if meta.RuntimeHandleID != "" { + dst.RuntimeHandleID = meta.RuntimeHandleID + } + if meta.RuntimeName != "" { + dst.RuntimeName = meta.RuntimeName + } + if meta.AgentSessionID != "" { + dst.AgentSessionID = meta.AgentSessionID + } + if meta.Prompt != "" { + dst.Prompt = meta.Prompt + } + return dst +} + // withMetadata attaches the separately-stored metadata to a record copy (a real // store would return them together). Caller holds s.mu. func (s *fakeStore) withMetadata(rec domain.SessionRecord) domain.SessionRecord { - if md := s.metadata[rec.ID]; len(md) > 0 { - rec.Metadata = cloneMap(md) - } + rec.Metadata = s.metadata[rec.ID] return rec } diff --git a/backend/internal/session/manager.go b/backend/internal/session/manager.go index e764f6a31..dce633055 100644 --- a/backend/internal/session/manager.go +++ b/backend/internal/session/manager.go @@ -19,7 +19,6 @@ import ( "time" "github.com/aoagents/agent-orchestrator/backend/internal/domain" - "github.com/aoagents/agent-orchestrator/backend/internal/lifecycle" "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) @@ -278,8 +277,8 @@ func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Sess // (the agent's id-capture path is a separate hook that may never have run, so // "no id" is the common case rather than an error). If neither is available // there is nothing to relaunch from — fail early, before any I/O. - agentSessionID := meta[lifecycle.MetaAgentSessionID] - seededPrompt := meta[lifecycle.MetaPrompt] + agentSessionID := meta.AgentSessionID + seededPrompt := meta.Prompt if agentSessionID == "" && seededPrompt == "" { return domain.Session{}, fmt.Errorf("restore %s: no agent session id or seeded prompt (cannot resume or relaunch)", id) } @@ -287,7 +286,7 @@ func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Sess ws, err := m.workspace.Restore(ctx, ports.WorkspaceConfig{ ProjectID: rec.ProjectID, SessionID: id, - Branch: meta[lifecycle.MetaBranch], + Branch: meta.Branch, }) if err != nil { return domain.Session{}, fmt.Errorf("restore %s: workspace restore: %w", id, err) @@ -335,7 +334,7 @@ func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Sess 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) } - if len(rec.Metadata) > 0 { + if !rec.Metadata.IsZero() { if revertErr := m.store.PatchMetadata(ctx, id, rec.Metadata); revertErr != nil { return domain.Session{}, fmt.Errorf("restore %s: revert metadata after spawn completed failure: %w (original error: %v)", id, revertErr, err) } @@ -440,17 +439,17 @@ func seedRecord(id domain.SessionID, cfg ports.SpawnConfig, now time.Time) domai // runtimeHandle / workspaceInfo reconstruct teardown handles from the metadata // the LCM persisted in OnSpawnCompleted (the metadata-key contract is shared // with the lifecycle package). -func runtimeHandle(meta map[string]string) ports.RuntimeHandle { +func runtimeHandle(meta domain.SessionMetadata) ports.RuntimeHandle { return ports.RuntimeHandle{ - ID: meta[lifecycle.MetaRuntimeHandleID], - RuntimeName: meta[lifecycle.MetaRuntimeName], + ID: meta.RuntimeHandleID, + RuntimeName: meta.RuntimeName, } } -func workspaceInfo(rec domain.SessionRecord, meta map[string]string) ports.WorkspaceInfo { +func workspaceInfo(rec domain.SessionRecord, meta domain.SessionMetadata) ports.WorkspaceInfo { return ports.WorkspaceInfo{ - Path: meta[lifecycle.MetaWorkspacePath], - Branch: meta[lifecycle.MetaBranch], + Path: meta.WorkspacePath, + Branch: meta.Branch, SessionID: rec.ID, ProjectID: rec.ProjectID, } diff --git a/backend/internal/session/manager_test.go b/backend/internal/session/manager_test.go index 5bb20d07b..c0c98cf7c 100644 --- a/backend/internal/session/manager_test.go +++ b/backend/internal/session/manager_test.go @@ -6,7 +6,6 @@ import ( "testing" "github.com/aoagents/agent-orchestrator/backend/internal/domain" - "github.com/aoagents/agent-orchestrator/backend/internal/lifecycle" "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) @@ -86,16 +85,15 @@ func TestSpawn_HappyPath(t *testing.T) { // persisted too so a later Restore that finds no captured agent session id // can still fall back to a fresh launch using the same prompt. meta, _ := h.store.GetMetadata(ctx, "sess-1") - for k, want := range map[string]string{ - lifecycle.MetaBranch: "feat/42", - lifecycle.MetaWorkspacePath: "/tmp/ws/sess-1", - lifecycle.MetaRuntimeHandleID: "rt-sess-1", - lifecycle.MetaRuntimeName: "tmux", - lifecycle.MetaPrompt: "do the thing\n\nbe careful", - } { - if meta[k] != want { - t.Errorf("meta[%q] = %q, want %q", k, meta[k], want) - } + want := domain.SessionMetadata{ + Branch: "feat/42", + WorkspacePath: "/tmp/ws/sess-1", + RuntimeHandleID: "rt-sess-1", + RuntimeName: "tmux", + Prompt: "do the thing\n\nbe careful", + } + if meta != want { + t.Errorf("metadata = %+v, want %+v", meta, want) } } @@ -300,7 +298,7 @@ func TestRestore_LiveSession_Rejected(t *testing.T) { } // The session is live (never torn down). Capture an agent id so the only thing // blocking restore is the non-terminal lifecycle, not missing metadata. - if err := h.store.PatchMetadata(ctx, "sess-1", map[string]string{lifecycle.MetaAgentSessionID: "agent-xyz"}); err != nil { + if err := h.store.PatchMetadata(ctx, "sess-1", domain.SessionMetadata{AgentSessionID: "agent-xyz"}); err != nil { t.Fatalf("patch metadata: %v", err) } createdBefore := len(h.runtime.created) @@ -398,7 +396,7 @@ func TestRestore_RelaunchesWithResumeCommand(t *testing.T) { t.Fatalf("kill: %v", err) } // The agent's resume id is captured in metadata (here set explicitly). - if err := h.store.PatchMetadata(ctx, "sess-1", map[string]string{lifecycle.MetaAgentSessionID: "agent-xyz"}); err != nil { + if err := h.store.PatchMetadata(ctx, "sess-1", domain.SessionMetadata{AgentSessionID: "agent-xyz"}); err != nil { t.Fatalf("patch metadata: %v", err) } @@ -505,7 +503,7 @@ func TestRestore_OnSpawnCompletedFailure_RollsBackRuntime(t *testing.T) { if _, err := h.sm.Kill(ctx, "sess-1", ports.KillOptions{Reason: ports.KillManual}); err != nil { t.Fatalf("kill: %v", err) } - if err := h.store.PatchMetadata(ctx, "sess-1", map[string]string{lifecycle.MetaAgentSessionID: "agent-xyz"}); err != nil { + if err := h.store.PatchMetadata(ctx, "sess-1", domain.SessionMetadata{AgentSessionID: "agent-xyz"}); err != nil { t.Fatalf("patch metadata: %v", err) } beforeMeta, _ := h.store.GetMetadata(ctx, "sess-1") @@ -528,7 +526,7 @@ func TestRestore_OnSpawnCompletedFailure_RollsBackRuntime(t *testing.T) { t.Fatalf("restore failure should advance revision twice, got %d want %d", rec.Lifecycle.Revision, before.Lifecycle.Revision+2) } afterMeta, _ := h.store.GetMetadata(ctx, "sess-1") - if !equalStringMap(afterMeta, beforeMeta) { + if afterMeta != beforeMeta { t.Fatalf("restore failure should restore metadata, got %+v want %+v", afterMeta, beforeMeta) } @@ -595,7 +593,7 @@ func seedTerminal(t *testing.T, h *harness, id domain.SessionID, wsPath string) }, 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 { + if err := h.store.PatchMetadata(ctx, id, domain.SessionMetadata{WorkspacePath: wsPath}); err != nil { t.Fatalf("patch metadata %s: %v", id, err) } } @@ -612,18 +610,6 @@ func equalStrings(a, b []string) bool { return true } -func equalStringMap(a, b map[string]string) bool { - if len(a) != len(b) { - return false - } - for k, v := range a { - if b[k] != v { - return false - } - } - return true -} - func contains(ids []domain.SessionID, id domain.SessionID) bool { for _, x := range ids { if x == id { diff --git a/backend/internal/storage/sqlite/gen/metadata.sql.go b/backend/internal/storage/sqlite/gen/metadata.sql.go index 96510eb80..2c0396f73 100644 --- a/backend/internal/storage/sqlite/gen/metadata.sql.go +++ b/backend/internal/storage/sqlite/gen/metadata.sql.go @@ -7,53 +7,76 @@ package gen import ( "context" + "time" ) -const getMetadata = `-- name: GetMetadata :many -SELECT key, value FROM session_metadata WHERE session_id = ? +const getSessionMetadata = `-- name: GetSessionMetadata :one +SELECT branch, workspace_path, runtime_handle_id, runtime_name, agent_session_id, prompt +FROM session_metadata +WHERE session_id = ? ` -type GetMetadataRow struct { - Key string - Value string +type GetSessionMetadataRow struct { + Branch string + WorkspacePath string + RuntimeHandleID string + RuntimeName string + AgentSessionID string + Prompt string } -func (q *Queries) GetMetadata(ctx context.Context, sessionID string) ([]GetMetadataRow, error) { - rows, err := q.db.QueryContext(ctx, getMetadata, sessionID) - if err != nil { - return nil, err - } - defer rows.Close() - items := []GetMetadataRow{} - for rows.Next() { - var i GetMetadataRow - if err := rows.Scan(&i.Key, &i.Value); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Close(); err != nil { - return nil, err - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil +func (q *Queries) GetSessionMetadata(ctx context.Context, sessionID string) (GetSessionMetadataRow, error) { + row := q.db.QueryRowContext(ctx, getSessionMetadata, sessionID) + var i GetSessionMetadataRow + err := row.Scan( + &i.Branch, + &i.WorkspacePath, + &i.RuntimeHandleID, + &i.RuntimeName, + &i.AgentSessionID, + &i.Prompt, + ) + return i, err } -const upsertMetadata = `-- name: UpsertMetadata :exec -INSERT INTO session_metadata (session_id, key, value) -VALUES (?, ?, ?) -ON CONFLICT (session_id, key) DO UPDATE SET value = excluded.value +const upsertSessionMetadata = `-- name: UpsertSessionMetadata :exec +INSERT INTO session_metadata ( + session_id, branch, workspace_path, runtime_handle_id, runtime_name, agent_session_id, prompt, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (session_id) DO UPDATE SET + branch = CASE WHEN excluded.branch <> '' THEN excluded.branch ELSE session_metadata.branch END, + workspace_path = CASE WHEN excluded.workspace_path <> '' THEN excluded.workspace_path ELSE session_metadata.workspace_path END, + runtime_handle_id = CASE WHEN excluded.runtime_handle_id <> '' THEN excluded.runtime_handle_id ELSE session_metadata.runtime_handle_id END, + runtime_name = CASE WHEN excluded.runtime_name <> '' THEN excluded.runtime_name ELSE session_metadata.runtime_name END, + agent_session_id = CASE WHEN excluded.agent_session_id <> '' THEN excluded.agent_session_id ELSE session_metadata.agent_session_id END, + prompt = CASE WHEN excluded.prompt <> '' THEN excluded.prompt ELSE session_metadata.prompt END, + updated_at = excluded.updated_at ` -type UpsertMetadataParams struct { - SessionID string - Key string - Value string +type UpsertSessionMetadataParams struct { + SessionID string + Branch string + WorkspacePath string + RuntimeHandleID string + RuntimeName string + AgentSessionID string + Prompt string + UpdatedAt time.Time } -func (q *Queries) UpsertMetadata(ctx context.Context, arg UpsertMetadataParams) error { - _, err := q.db.ExecContext(ctx, upsertMetadata, arg.SessionID, arg.Key, arg.Value) +// Merge semantics: an empty incoming column is "leave unchanged", so a partial +// patch (e.g. spawn writing only the runtime handle) never clobbers a value set +// earlier (e.g. the branch set at creation). Mirrors the old per-key map merge. +func (q *Queries) UpsertSessionMetadata(ctx context.Context, arg UpsertSessionMetadataParams) error { + _, err := q.db.ExecContext(ctx, upsertSessionMetadata, + arg.SessionID, + arg.Branch, + arg.WorkspacePath, + arg.RuntimeHandleID, + arg.RuntimeName, + arg.AgentSessionID, + arg.Prompt, + arg.UpdatedAt, + ) return err } diff --git a/backend/internal/storage/sqlite/gen/models.go b/backend/internal/storage/sqlite/gen/models.go index dccf25c46..339062bf0 100644 --- a/backend/internal/storage/sqlite/gen/models.go +++ b/backend/internal/storage/sqlite/gen/models.go @@ -34,14 +34,34 @@ type Outbox struct { CreatedAt time.Time } -type PrEnrichment struct { - SessionID string - CiSummary string - ReviewDecision string - Mergeability string - PendingComments string - CiLogTail string - LastFetchedAt time.Time +type Pr struct { + SessionID string + ReviewDecision string + Mergeability string + CiState string + CiPassed int64 + CiFailed int64 + CiPending int64 + CiLogTail string + LastFetchedAt time.Time +} + +type PrCheck struct { + SessionID string + Name string + Status string + Url string +} + +type PrComment struct { + SessionID string + CommentID string + Author string + File string + Line int64 + Body string + Resolved int64 + CreatedAt time.Time } type Project struct { @@ -93,7 +113,12 @@ type Session struct { } type SessionMetadatum struct { - SessionID string - Key string - Value string + SessionID string + Branch string + WorkspacePath string + RuntimeHandleID string + RuntimeName string + AgentSessionID string + Prompt string + UpdatedAt time.Time } diff --git a/backend/internal/storage/sqlite/gen/pr.sql.go b/backend/internal/storage/sqlite/gen/pr.sql.go new file mode 100644 index 000000000..95cbd20ae --- /dev/null +++ b/backend/internal/storage/sqlite/gen/pr.sql.go @@ -0,0 +1,235 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: pr.sql + +package gen + +import ( + "context" + "time" +) + +const deletePR = `-- name: DeletePR :exec +DELETE FROM pr WHERE session_id = ? +` + +func (q *Queries) DeletePR(ctx context.Context, sessionID string) error { + _, err := q.db.ExecContext(ctx, deletePR, sessionID) + return err +} + +const deletePRChecks = `-- name: DeletePRChecks :exec +DELETE FROM pr_check WHERE session_id = ? +` + +func (q *Queries) DeletePRChecks(ctx context.Context, sessionID string) error { + _, err := q.db.ExecContext(ctx, deletePRChecks, sessionID) + return err +} + +const deletePRComments = `-- name: DeletePRComments :exec +DELETE FROM pr_comment WHERE session_id = ? +` + +func (q *Queries) DeletePRComments(ctx context.Context, sessionID string) error { + _, err := q.db.ExecContext(ctx, deletePRComments, sessionID) + return err +} + +const getPR = `-- name: GetPR :one +SELECT session_id, review_decision, mergeability, ci_state, ci_passed, ci_failed, ci_pending, ci_log_tail, last_fetched_at +FROM pr +WHERE session_id = ? +` + +func (q *Queries) GetPR(ctx context.Context, sessionID string) (Pr, error) { + row := q.db.QueryRowContext(ctx, getPR, sessionID) + var i Pr + err := row.Scan( + &i.SessionID, + &i.ReviewDecision, + &i.Mergeability, + &i.CiState, + &i.CiPassed, + &i.CiFailed, + &i.CiPending, + &i.CiLogTail, + &i.LastFetchedAt, + ) + return i, err +} + +const insertPRCheck = `-- name: InsertPRCheck :exec +INSERT INTO pr_check (session_id, name, status, url) VALUES (?, ?, ?, ?) +` + +type InsertPRCheckParams struct { + SessionID string + Name string + Status string + Url string +} + +func (q *Queries) InsertPRCheck(ctx context.Context, arg InsertPRCheckParams) error { + _, err := q.db.ExecContext(ctx, insertPRCheck, + arg.SessionID, + arg.Name, + arg.Status, + arg.Url, + ) + return err +} + +const insertPRComment = `-- name: InsertPRComment :exec +INSERT INTO pr_comment (session_id, comment_id, author, file, line, body, resolved, created_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?) +` + +type InsertPRCommentParams struct { + SessionID string + CommentID string + Author string + File string + Line int64 + Body string + Resolved int64 + CreatedAt time.Time +} + +func (q *Queries) InsertPRComment(ctx context.Context, arg InsertPRCommentParams) error { + _, err := q.db.ExecContext(ctx, insertPRComment, + arg.SessionID, + arg.CommentID, + arg.Author, + arg.File, + arg.Line, + arg.Body, + arg.Resolved, + arg.CreatedAt, + ) + return err +} + +const listPRChecks = `-- name: ListPRChecks :many +SELECT name, status, url FROM pr_check WHERE session_id = ? ORDER BY name +` + +type ListPRChecksRow struct { + Name string + Status string + Url string +} + +func (q *Queries) ListPRChecks(ctx context.Context, sessionID string) ([]ListPRChecksRow, error) { + rows, err := q.db.QueryContext(ctx, listPRChecks, sessionID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListPRChecksRow{} + for rows.Next() { + var i ListPRChecksRow + if err := rows.Scan(&i.Name, &i.Status, &i.Url); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listPRComments = `-- name: ListPRComments :many +SELECT comment_id, author, file, line, body, resolved, created_at +FROM pr_comment +WHERE session_id = ? +ORDER BY created_at, comment_id +` + +type ListPRCommentsRow struct { + CommentID string + Author string + File string + Line int64 + Body string + Resolved int64 + CreatedAt time.Time +} + +func (q *Queries) ListPRComments(ctx context.Context, sessionID string) ([]ListPRCommentsRow, error) { + rows, err := q.db.QueryContext(ctx, listPRComments, sessionID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListPRCommentsRow{} + for rows.Next() { + var i ListPRCommentsRow + if err := rows.Scan( + &i.CommentID, + &i.Author, + &i.File, + &i.Line, + &i.Body, + &i.Resolved, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const upsertPR = `-- name: UpsertPR :exec +INSERT INTO pr ( + session_id, review_decision, mergeability, ci_state, ci_passed, ci_failed, ci_pending, ci_log_tail, last_fetched_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (session_id) DO UPDATE SET + review_decision = excluded.review_decision, + mergeability = excluded.mergeability, + ci_state = excluded.ci_state, + ci_passed = excluded.ci_passed, + ci_failed = excluded.ci_failed, + ci_pending = excluded.ci_pending, + ci_log_tail = excluded.ci_log_tail, + last_fetched_at = excluded.last_fetched_at +` + +type UpsertPRParams struct { + SessionID string + ReviewDecision string + Mergeability string + CiState string + CiPassed int64 + CiFailed int64 + CiPending int64 + CiLogTail string + LastFetchedAt time.Time +} + +func (q *Queries) UpsertPR(ctx context.Context, arg UpsertPRParams) error { + _, err := q.db.ExecContext(ctx, upsertPR, + arg.SessionID, + arg.ReviewDecision, + arg.Mergeability, + arg.CiState, + arg.CiPassed, + arg.CiFailed, + arg.CiPending, + arg.CiLogTail, + arg.LastFetchedAt, + ) + return err +} diff --git a/backend/internal/storage/sqlite/gen/pr_enrichment.sql.go b/backend/internal/storage/sqlite/gen/pr_enrichment.sql.go deleted file mode 100644 index c0643104b..000000000 --- a/backend/internal/storage/sqlite/gen/pr_enrichment.sql.go +++ /dev/null @@ -1,76 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: pr_enrichment.sql - -package gen - -import ( - "context" - "time" -) - -const deletePREnrichment = `-- name: DeletePREnrichment :exec -DELETE FROM pr_enrichment WHERE session_id = ? -` - -func (q *Queries) DeletePREnrichment(ctx context.Context, sessionID string) error { - _, err := q.db.ExecContext(ctx, deletePREnrichment, sessionID) - return err -} - -const getPREnrichment = `-- name: GetPREnrichment :one -SELECT session_id, ci_summary, review_decision, mergeability, pending_comments, ci_log_tail, last_fetched_at -FROM pr_enrichment -WHERE session_id = ? -` - -func (q *Queries) GetPREnrichment(ctx context.Context, sessionID string) (PrEnrichment, error) { - row := q.db.QueryRowContext(ctx, getPREnrichment, sessionID) - var i PrEnrichment - err := row.Scan( - &i.SessionID, - &i.CiSummary, - &i.ReviewDecision, - &i.Mergeability, - &i.PendingComments, - &i.CiLogTail, - &i.LastFetchedAt, - ) - return i, err -} - -const upsertPREnrichment = `-- name: UpsertPREnrichment :exec -INSERT INTO pr_enrichment (session_id, ci_summary, review_decision, mergeability, pending_comments, ci_log_tail, last_fetched_at) -VALUES (?, ?, ?, ?, ?, ?, ?) -ON CONFLICT (session_id) DO UPDATE SET - ci_summary = excluded.ci_summary, - review_decision = excluded.review_decision, - mergeability = excluded.mergeability, - pending_comments = excluded.pending_comments, - ci_log_tail = excluded.ci_log_tail, - last_fetched_at = excluded.last_fetched_at -` - -type UpsertPREnrichmentParams struct { - SessionID string - CiSummary string - ReviewDecision string - Mergeability string - PendingComments string - CiLogTail string - LastFetchedAt time.Time -} - -func (q *Queries) UpsertPREnrichment(ctx context.Context, arg UpsertPREnrichmentParams) error { - _, err := q.db.ExecContext(ctx, upsertPREnrichment, - arg.SessionID, - arg.CiSummary, - arg.ReviewDecision, - arg.Mergeability, - arg.PendingComments, - arg.CiLogTail, - arg.LastFetchedAt, - ) - return err -} diff --git a/backend/internal/storage/sqlite/gen/querier.go b/backend/internal/storage/sqlite/gen/querier.go index 76dd1aab5..83aa0c7ed 100644 --- a/backend/internal/storage/sqlite/gen/querier.go +++ b/backend/internal/storage/sqlite/gen/querier.go @@ -10,25 +10,31 @@ import ( type Querier interface { ArchiveProject(ctx context.Context, arg ArchiveProjectParams) error - DeletePREnrichment(ctx context.Context, sessionID string) error + DeletePR(ctx context.Context, sessionID string) error + DeletePRChecks(ctx context.Context, sessionID string) error + DeletePRComments(ctx context.Context, sessionID string) error DeleteProject(ctx context.Context, id string) error DeleteReactionTracker(ctx context.Context, arg DeleteReactionTrackerParams) error DeleteSentOutboxBelow(ctx context.Context, changeLogSeq int64) (int64, error) DeleteSessionReactionTrackers(ctx context.Context, sessionID string) error GetConsumerOffset(ctx context.Context, consumer string) (int64, error) - GetMetadata(ctx context.Context, sessionID string) ([]GetMetadataRow, error) - GetPREnrichment(ctx context.Context, sessionID string) (PrEnrichment, error) + GetPR(ctx context.Context, sessionID string) (Pr, error) GetProject(ctx context.Context, id string) (Project, error) GetSession(ctx context.Context, id string) (Session, error) + GetSessionMetadata(ctx context.Context, sessionID string) (GetSessionMetadataRow, error) GetSessionRevision(ctx context.Context, id string) (int64, error) // Appends a canonical-write record and returns its monotonic seq so the same // transaction can thread it into the outbox row. InsertChangeLog(ctx context.Context, arg InsertChangeLogParams) (int64, error) InsertOutbox(ctx context.Context, arg InsertOutboxParams) error + InsertPRCheck(ctx context.Context, arg InsertPRCheckParams) error + InsertPRComment(ctx context.Context, arg InsertPRCommentParams) error // CAS insert: only succeeds for a brand-new id. Incoming revision must be 0; // the row is persisted at revision 1. InsertSession(ctx context.Context, arg InsertSessionParams) (int64, error) ListAllSessions(ctx context.Context) ([]Session, error) + ListPRChecks(ctx context.Context, sessionID string) ([]ListPRChecksRow, error) + ListPRComments(ctx context.Context, sessionID string) ([]ListPRCommentsRow, error) ListProjects(ctx context.Context) ([]Project, error) ListReactionTrackers(ctx context.Context) ([]ReactionTracker, error) ListSessionsByProject(ctx context.Context, projectID string) ([]Session, error) @@ -41,10 +47,13 @@ type Querier interface { // revision (@expected_revision). 0 rows affected => revision mismatch. UpdateSessionCAS(ctx context.Context, arg UpdateSessionCASParams) (int64, error) UpsertConsumerOffset(ctx context.Context, arg UpsertConsumerOffsetParams) error - UpsertMetadata(ctx context.Context, arg UpsertMetadataParams) error - UpsertPREnrichment(ctx context.Context, arg UpsertPREnrichmentParams) error + UpsertPR(ctx context.Context, arg UpsertPRParams) error UpsertProject(ctx context.Context, arg UpsertProjectParams) error UpsertReactionTracker(ctx context.Context, arg UpsertReactionTrackerParams) error + // Merge semantics: an empty incoming column is "leave unchanged", so a partial + // patch (e.g. spawn writing only the runtime handle) never clobbers a value set + // earlier (e.g. the branch set at creation). Mirrors the old per-key map merge. + UpsertSessionMetadata(ctx context.Context, arg UpsertSessionMetadataParams) error } var _ Querier = (*Queries)(nil) diff --git a/backend/internal/storage/sqlite/migrations/0001_init.sql b/backend/internal/storage/sqlite/migrations/0001_init.sql index f343e16d1..382241256 100644 --- a/backend/internal/storage/sqlite/migrations/0001_init.sql +++ b/backend/internal/storage/sqlite/migrations/0001_init.sql @@ -39,14 +39,21 @@ CREATE TABLE sessions ( CREATE INDEX idx_sessions_project ON sessions (project_id); --- session_metadata is the opaque key/value side-channel (branch, workspacePath, --- runtimeHandleId, runtimeName, agentSessionId, prompt). Written by --- PatchMetadata; never bumps revision and never emits a CDC event. +-- session_metadata is the 1:1 typed side-channel for a session's operational +-- handles and seed inputs — the fields the Session Manager and reaper need but +-- that are NOT part of the canonical lifecycle. One row per session, named +-- columns (not a free-form key/value bag), so the set of metadata a session can +-- carry is fixed by the schema. Written by PatchMetadata; never bumps revision +-- and never emits a CDC event. CREATE TABLE session_metadata ( - session_id TEXT NOT NULL REFERENCES sessions (id) ON DELETE CASCADE, - key TEXT NOT NULL, - value TEXT NOT NULL, - PRIMARY KEY (session_id, key) + session_id TEXT PRIMARY KEY REFERENCES sessions (id) ON DELETE CASCADE, + branch TEXT NOT NULL DEFAULT '', + workspace_path TEXT NOT NULL DEFAULT '', + runtime_handle_id TEXT NOT NULL DEFAULT '', + runtime_name TEXT NOT NULL DEFAULT '', + agent_session_id TEXT NOT NULL DEFAULT '', + prompt TEXT NOT NULL DEFAULT '', + updated_at TIMESTAMP NOT NULL ); -- change_log is the durable, ordered record of every canonical write. seq is the diff --git a/backend/internal/storage/sqlite/migrations/0002_pr_projects.sql b/backend/internal/storage/sqlite/migrations/0002_pr_projects.sql index 4421f0ddc..da987ed5c 100644 --- a/backend/internal/storage/sqlite/migrations/0002_pr_projects.sql +++ b/backend/internal/storage/sqlite/migrations/0002_pr_projects.sql @@ -26,25 +26,60 @@ CREATE TABLE projects ( archived_at TIMESTAMP ); --- pr_enrichment is the SCM observer's per-session cache of the rich PR facts that --- do NOT live in the canonical lifecycle (which keeps only pr_state/reason/number/ --- url). It is 1:1 with a session (a PR is always tied to a session by its branch), --- written by the SCM observer OFF the canonical CDC path (no revision bump, no --- change_log/outbox event), and cascades away with its session. -CREATE TABLE pr_enrichment ( +-- pr is the SCM observer's per-session cache of the rich PR facts that do NOT +-- live in the canonical lifecycle (which keeps only pr_state/reason/number/url). +-- 1:1 with a session (a PR is tied to a session by its branch), written by the +-- SCM observer OFF the canonical CDC path (no revision bump, no change_log/outbox +-- event), and cascades away with its session. Scalar facts are typed columns — +-- review_decision/mergeability/ci_state are CHECK-constrained enums and the CI +-- counts are integers, not opaque strings; the list facts (individual checks and +-- review comments) are normalized into pr_check / pr_comment. +CREATE TABLE pr ( session_id TEXT PRIMARY KEY REFERENCES sessions (id) ON DELETE CASCADE, - ci_summary TEXT NOT NULL DEFAULT '', - review_decision TEXT NOT NULL DEFAULT '', - mergeability TEXT NOT NULL DEFAULT '', - pending_comments TEXT NOT NULL DEFAULT '', + review_decision TEXT NOT NULL DEFAULT 'none' + CHECK (review_decision IN ('none', 'approved', 'changes_requested', 'review_required')), + mergeability TEXT NOT NULL DEFAULT 'unknown' + CHECK (mergeability IN ('unknown', 'mergeable', 'conflicting', 'blocked', 'unstable')), + ci_state TEXT NOT NULL DEFAULT 'unknown' + CHECK (ci_state IN ('unknown', 'pending', 'passing', 'failing')), + ci_passed INTEGER NOT NULL DEFAULT 0, + ci_failed INTEGER NOT NULL DEFAULT 0, + ci_pending INTEGER NOT NULL DEFAULT 0, ci_log_tail TEXT NOT NULL DEFAULT '', last_fetched_at TIMESTAMP NOT NULL ); +-- pr_check is one CI check belonging to a pr (the normalized form of the old +-- ci_summary string). It cascades from pr, so it cannot outlive its PR facts. +CREATE TABLE pr_check ( + session_id TEXT NOT NULL REFERENCES pr (session_id) ON DELETE CASCADE, + name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'unknown' + CHECK (status IN ('unknown', 'queued', 'in_progress', 'passed', 'failed', 'skipped', 'cancelled')), + url TEXT NOT NULL DEFAULT '', + PRIMARY KEY (session_id, name) +); + +-- pr_comment is one unresolved review comment belonging to a pr (the normalized +-- form of the old pending_comments JSON-in-a-string). Cascades from pr. +CREATE TABLE pr_comment ( + session_id TEXT NOT NULL REFERENCES pr (session_id) ON DELETE CASCADE, + comment_id TEXT NOT NULL, + author TEXT NOT NULL DEFAULT '', + file TEXT NOT NULL DEFAULT '', + line INTEGER NOT NULL DEFAULT 0, + body TEXT NOT NULL DEFAULT '', + resolved INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMP NOT NULL, + PRIMARY KEY (session_id, comment_id) +); + -- +goose StatementEnd -- +goose Down -- +goose StatementBegin -DROP TABLE pr_enrichment; +DROP TABLE pr_comment; +DROP TABLE pr_check; +DROP TABLE pr; DROP TABLE projects; -- +goose StatementEnd diff --git a/backend/internal/storage/sqlite/pr_projects_test.go b/backend/internal/storage/sqlite/pr_projects_test.go index 6cdd20bc1..58227b1f3 100644 --- a/backend/internal/storage/sqlite/pr_projects_test.go +++ b/backend/internal/storage/sqlite/pr_projects_test.go @@ -2,6 +2,7 @@ package sqlite import ( "context" + "reflect" "testing" "time" @@ -88,41 +89,122 @@ func TestArchiveProjectHidesFromListButGetResolves(t *testing.T) { } } -func TestPREnrichmentUpsertGetDelete(t *testing.T) { +func TestPRUpsertGetDelete(t *testing.T) { s := newTestStore(t) ctx := context.Background() now := time.Now().UTC().Truncate(time.Second) - // pr_enrichment FKs sessions(id); seed the session first. + // pr FKs sessions(id); seed the session first. if err := s.Upsert(ctx, sampleRecord("s1"), ports.EventSessionCreated); err != nil { t.Fatalf("seed session: %v", err) } - if _, ok, err := s.GetPREnrichment(ctx, "s1"); err != nil || ok { + if _, ok, err := s.GetPR(ctx, "s1"); err != nil || ok { t.Fatalf("get missing: ok=%v err=%v", ok, err) } - e := PREnrichmentRow{ - SessionID: "s1", CISummary: "3 passing, 1 failing", ReviewDecision: "changes_requested", - Mergeability: "blocked", PendingComments: `[{"path":"a.go"}]`, CILogTail: "FAIL TestX", + pr := PRRow{ + SessionID: "s1", ReviewDecision: "changes_requested", Mergeability: "blocked", + CIState: "failing", CIPassed: 3, CIFailed: 1, CIPending: 0, CILogTail: "FAIL TestX", LastFetchedAt: now, } - if err := s.UpsertPREnrichment(ctx, e); err != nil { + if err := s.UpsertPR(ctx, pr); err != nil { t.Fatalf("upsert: %v", err) } - got, ok, err := s.GetPREnrichment(ctx, "s1") + got, ok, err := s.GetPR(ctx, "s1") if err != nil || !ok { t.Fatalf("get: ok=%v err=%v", ok, err) } - if got != e { - t.Fatalf("round-trip mismatch:\n got %+v\nwant %+v", got, e) + if got != pr { + t.Fatalf("round-trip mismatch:\n got %+v\nwant %+v", got, pr) } - if err := s.DeletePREnrichment(ctx, "s1"); err != nil { + if err := s.DeletePR(ctx, "s1"); err != nil { t.Fatalf("delete: %v", err) } - if _, ok, _ := s.GetPREnrichment(ctx, "s1"); ok { - t.Fatal("enrichment should be gone after delete") + if _, ok, _ := s.GetPR(ctx, "s1"); ok { + t.Fatal("pr should be gone after delete") + } +} + +func TestPRRejectsBadEnum(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + if err := s.Upsert(ctx, sampleRecord("s1"), ports.EventSessionCreated); err != nil { + t.Fatalf("seed session: %v", err) + } + // review_decision is a CHECK-constrained enum; an off-list value must fail. + err := s.UpsertPR(ctx, PRRow{ + SessionID: "s1", ReviewDecision: "definitely_not_a_decision", + Mergeability: "unknown", CIState: "unknown", LastFetchedAt: time.Now().UTC(), + }) + if err == nil { + t.Fatal("expected CHECK constraint to reject an invalid review_decision") + } +} + +func TestPRChecksAndCommentsReplaceAndList(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Second) + + if err := s.Upsert(ctx, sampleRecord("s1"), ports.EventSessionCreated); err != nil { + t.Fatalf("seed session: %v", err) + } + // pr_check / pr_comment FK pr(session_id); the pr row must exist first. + if err := s.UpsertPR(ctx, PRRow{ + SessionID: "s1", ReviewDecision: "review_required", Mergeability: "unknown", + CIState: "pending", LastFetchedAt: now, + }); err != nil { + t.Fatalf("upsert pr: %v", err) + } + + checks := []PRCheck{ + {Name: "build", Status: "passed", URL: "https://ci/build"}, + {Name: "test", Status: "failed", URL: "https://ci/test"}, + } + if err := s.ReplacePRChecks(ctx, "s1", checks); err != nil { + t.Fatalf("replace checks: %v", err) + } + gotChecks, err := s.ListPRChecks(ctx, "s1") + if err != nil { + t.Fatalf("list checks: %v", err) + } + if !reflect.DeepEqual(gotChecks, checks) { + t.Fatalf("checks = %+v, want %+v", gotChecks, checks) + } + // Replace is a set-replace, not a merge: a shorter set removes the rest. + if err := s.ReplacePRChecks(ctx, "s1", []PRCheck{{Name: "build", Status: "passed"}}); err != nil { + t.Fatalf("replace checks 2: %v", err) + } + if gotChecks, _ = s.ListPRChecks(ctx, "s1"); len(gotChecks) != 1 { + t.Fatalf("after replace, checks = %+v, want 1", gotChecks) + } + + comments := []PRComment{ + {CommentID: "c1", Author: "alice", File: "a.go", Line: 10, Body: "nit", Resolved: false, CreatedAt: now}, + {CommentID: "c2", Author: "bob", File: "b.go", Line: 20, Body: "bug", Resolved: true, CreatedAt: now.Add(time.Second)}, + } + if err := s.ReplacePRComments(ctx, "s1", comments); err != nil { + t.Fatalf("replace comments: %v", err) + } + gotComments, err := s.ListPRComments(ctx, "s1") + if err != nil { + t.Fatalf("list comments: %v", err) + } + if !reflect.DeepEqual(gotComments, comments) { + t.Fatalf("comments = %+v, want %+v", gotComments, comments) + } + + // Deleting the pr cascades its checks and comments. + if err := s.DeletePR(ctx, "s1"); err != nil { + t.Fatalf("delete pr: %v", err) + } + if c, _ := s.ListPRChecks(ctx, "s1"); len(c) != 0 { + t.Fatalf("checks not cascaded: %+v", c) + } + if c, _ := s.ListPRComments(ctx, "s1"); len(c) != 0 { + t.Fatalf("comments not cascaded: %+v", c) } } diff --git a/backend/internal/storage/sqlite/pr_store.go b/backend/internal/storage/sqlite/pr_store.go index 70efb7ce7..c7d436bda 100644 --- a/backend/internal/storage/sqlite/pr_store.go +++ b/backend/internal/storage/sqlite/pr_store.go @@ -10,57 +10,185 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen" ) -// PREnrichmentRow is the SCM observer's cache of the rich PR facts that do not -// live in the canonical lifecycle (which keeps only pr_state/reason/number/url). -// It is 1:1 with a session and written OFF the canonical CDC path: upserting it -// never bumps revision and never emits a change_log/outbox event. pending_comments -// and ci_log_tail are opaque blobs the SCM observer serializes. -type PREnrichmentRow struct { - SessionID string - CISummary string - ReviewDecision string - Mergeability string - PendingComments string - CILogTail string - LastFetchedAt time.Time +// PRRow is the SCM observer's cache of the scalar PR facts that do not live in +// the canonical lifecycle (which keeps only pr_state/reason/number/url). It is +// 1:1 with a session and written OFF the canonical CDC path: upserting it never +// bumps revision and never emits a change_log/outbox event. The list facts +// (checks, comments) are separate rows — see PRCheck / PRComment. +type PRRow struct { + SessionID string + ReviewDecision string // none | approved | changes_requested | review_required + Mergeability string // unknown | mergeable | conflicting | blocked | unstable + CIState string // unknown | pending | passing | failing + CIPassed int64 + CIFailed int64 + CIPending int64 + CILogTail string + LastFetchedAt time.Time } -// UpsertPREnrichment inserts or replaces the cached PR facts for one session. -func (s *Store) UpsertPREnrichment(ctx context.Context, r PREnrichmentRow) error { - return s.q.UpsertPREnrichment(ctx, gen.UpsertPREnrichmentParams{ - SessionID: r.SessionID, - CiSummary: r.CISummary, - ReviewDecision: r.ReviewDecision, - Mergeability: r.Mergeability, - PendingComments: r.PendingComments, - CiLogTail: r.CILogTail, - LastFetchedAt: r.LastFetchedAt, +// PRCheck is one CI check belonging to a session's PR. +type PRCheck struct { + Name string + Status string // unknown | queued | in_progress | passed | failed | skipped | cancelled + URL string +} + +// PRComment is one review comment belonging to a session's PR. +type PRComment struct { + CommentID string + Author string + File string + Line int64 + Body string + Resolved bool + CreatedAt time.Time +} + +// UpsertPR inserts or replaces the scalar PR facts for one session. +func (s *Store) UpsertPR(ctx context.Context, r PRRow) error { + return s.q.UpsertPR(ctx, gen.UpsertPRParams{ + SessionID: r.SessionID, + ReviewDecision: r.ReviewDecision, + Mergeability: r.Mergeability, + CiState: r.CIState, + CiPassed: r.CIPassed, + CiFailed: r.CIFailed, + CiPending: r.CIPending, + CiLogTail: r.CILogTail, + LastFetchedAt: r.LastFetchedAt, }) } -// GetPREnrichment returns the cached PR facts for one session. ok is false when -// no row exists (the SCM observer has not yet fetched, or the session has no PR). -func (s *Store) GetPREnrichment(ctx context.Context, sessionID string) (PREnrichmentRow, bool, error) { - e, err := s.q.GetPREnrichment(ctx, sessionID) +// GetPR returns the scalar PR facts for one session. ok is false when no row +// exists (the SCM observer has not fetched yet, or the session has no PR). +func (s *Store) GetPR(ctx context.Context, sessionID string) (PRRow, bool, error) { + p, err := s.q.GetPR(ctx, sessionID) if errors.Is(err, sql.ErrNoRows) { - return PREnrichmentRow{}, false, nil + return PRRow{}, false, nil } if err != nil { - return PREnrichmentRow{}, false, fmt.Errorf("get pr enrichment: %w", err) + return PRRow{}, false, fmt.Errorf("get pr: %w", err) } - return PREnrichmentRow{ - SessionID: e.SessionID, - CISummary: e.CiSummary, - ReviewDecision: e.ReviewDecision, - Mergeability: e.Mergeability, - PendingComments: e.PendingComments, - CILogTail: e.CiLogTail, - LastFetchedAt: e.LastFetchedAt, + return PRRow{ + SessionID: p.SessionID, + ReviewDecision: p.ReviewDecision, + Mergeability: p.Mergeability, + CIState: p.CiState, + CIPassed: p.CiPassed, + CIFailed: p.CiFailed, + CIPending: p.CiPending, + CILogTail: p.CiLogTail, + LastFetchedAt: p.LastFetchedAt, }, true, nil } -// DeletePREnrichment drops the cached PR facts for one session. Normally -// unnecessary (the FK cascades on session delete), exposed for explicit eviction. -func (s *Store) DeletePREnrichment(ctx context.Context, sessionID string) error { - return s.q.DeletePREnrichment(ctx, sessionID) +// DeletePR drops the scalar PR facts for one session, cascading its checks and +// comments. Normally unnecessary (the chain cascades on session delete); exposed +// for explicit eviction. +func (s *Store) DeletePR(ctx context.Context, sessionID string) error { + return s.q.DeletePR(ctx, sessionID) +} + +// ReplacePRChecks atomically replaces the full set of CI checks for a session's +// PR — each SCM fetch reports the current set, so a replace (not a merge) keeps +// the table in sync (a check that disappeared upstream is removed). The PR row +// must already exist (pr_check FKs pr). +func (s *Store) ReplacePRChecks(ctx context.Context, sessionID string, checks []PRCheck) error { + return s.inTx(ctx, "replace pr checks", func(qtx *gen.Queries) error { + if err := qtx.DeletePRChecks(ctx, sessionID); err != nil { + return err + } + for _, c := range checks { + if err := qtx.InsertPRCheck(ctx, gen.InsertPRCheckParams{ + SessionID: sessionID, + Name: c.Name, + Status: c.Status, + Url: c.URL, + }); err != nil { + return fmt.Errorf("check %q: %w", c.Name, err) + } + } + return nil + }) +} + +// ListPRChecks returns a session's CI checks, ordered by name. +func (s *Store) ListPRChecks(ctx context.Context, sessionID string) ([]PRCheck, error) { + rows, err := s.q.ListPRChecks(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("list pr checks: %w", err) + } + out := make([]PRCheck, 0, len(rows)) + for _, r := range rows { + out = append(out, PRCheck{Name: r.Name, Status: r.Status, URL: r.Url}) + } + return out, nil +} + +// ReplacePRComments atomically replaces the full set of review comments for a +// session's PR (same replace-not-merge rationale as ReplacePRChecks). +func (s *Store) ReplacePRComments(ctx context.Context, sessionID string, comments []PRComment) error { + return s.inTx(ctx, "replace pr comments", func(qtx *gen.Queries) error { + if err := qtx.DeletePRComments(ctx, sessionID); err != nil { + return err + } + for _, c := range comments { + if err := qtx.InsertPRComment(ctx, gen.InsertPRCommentParams{ + SessionID: sessionID, + CommentID: c.CommentID, + Author: c.Author, + File: c.File, + Line: c.Line, + Body: c.Body, + Resolved: boolToInt(c.Resolved), + CreatedAt: c.CreatedAt, + }); err != nil { + return fmt.Errorf("comment %q: %w", c.CommentID, err) + } + } + return nil + }) +} + +// ListPRComments returns a session's review comments, ordered by creation time. +func (s *Store) ListPRComments(ctx context.Context, sessionID string) ([]PRComment, error) { + rows, err := s.q.ListPRComments(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("list pr comments: %w", err) + } + out := make([]PRComment, 0, len(rows)) + for _, r := range rows { + out = append(out, PRComment{ + CommentID: r.CommentID, + Author: r.Author, + File: r.File, + Line: r.Line, + Body: r.Body, + Resolved: r.Resolved != 0, + CreatedAt: r.CreatedAt, + }) + } + return out, nil +} + +// inTx runs fn inside a single transaction over the store's queries, rolling +// back on error. +func (s *Store) inTx(ctx context.Context, what string, fn func(*gen.Queries) error) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin %s: %w", what, err) + } + defer tx.Rollback() + if err := fn(s.q.WithTx(tx)); err != nil { + return fmt.Errorf("%s: %w", what, err) + } + return tx.Commit() +} + +func boolToInt(b bool) int64 { + if b { + return 1 + } + return 0 } diff --git a/backend/internal/storage/sqlite/queries/metadata.sql b/backend/internal/storage/sqlite/queries/metadata.sql index 45079bb25..158552daa 100644 --- a/backend/internal/storage/sqlite/queries/metadata.sql +++ b/backend/internal/storage/sqlite/queries/metadata.sql @@ -1,7 +1,20 @@ --- name: GetMetadata :many -SELECT key, value FROM session_metadata WHERE session_id = ?; +-- name: GetSessionMetadata :one +SELECT branch, workspace_path, runtime_handle_id, runtime_name, agent_session_id, prompt +FROM session_metadata +WHERE session_id = ?; --- name: UpsertMetadata :exec -INSERT INTO session_metadata (session_id, key, value) -VALUES (?, ?, ?) -ON CONFLICT (session_id, key) DO UPDATE SET value = excluded.value; +-- name: UpsertSessionMetadata :exec +-- Merge semantics: an empty incoming column is "leave unchanged", so a partial +-- patch (e.g. spawn writing only the runtime handle) never clobbers a value set +-- earlier (e.g. the branch set at creation). Mirrors the old per-key map merge. +INSERT INTO session_metadata ( + session_id, branch, workspace_path, runtime_handle_id, runtime_name, agent_session_id, prompt, updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (session_id) DO UPDATE SET + branch = CASE WHEN excluded.branch <> '' THEN excluded.branch ELSE session_metadata.branch END, + workspace_path = CASE WHEN excluded.workspace_path <> '' THEN excluded.workspace_path ELSE session_metadata.workspace_path END, + runtime_handle_id = CASE WHEN excluded.runtime_handle_id <> '' THEN excluded.runtime_handle_id ELSE session_metadata.runtime_handle_id END, + runtime_name = CASE WHEN excluded.runtime_name <> '' THEN excluded.runtime_name ELSE session_metadata.runtime_name END, + agent_session_id = CASE WHEN excluded.agent_session_id <> '' THEN excluded.agent_session_id ELSE session_metadata.agent_session_id END, + prompt = CASE WHEN excluded.prompt <> '' THEN excluded.prompt ELSE session_metadata.prompt END, + updated_at = excluded.updated_at; diff --git a/backend/internal/storage/sqlite/queries/pr.sql b/backend/internal/storage/sqlite/queries/pr.sql new file mode 100644 index 000000000..13c14a78d --- /dev/null +++ b/backend/internal/storage/sqlite/queries/pr.sql @@ -0,0 +1,43 @@ +-- name: UpsertPR :exec +INSERT INTO pr ( + session_id, review_decision, mergeability, ci_state, ci_passed, ci_failed, ci_pending, ci_log_tail, last_fetched_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (session_id) DO UPDATE SET + review_decision = excluded.review_decision, + mergeability = excluded.mergeability, + ci_state = excluded.ci_state, + ci_passed = excluded.ci_passed, + ci_failed = excluded.ci_failed, + ci_pending = excluded.ci_pending, + ci_log_tail = excluded.ci_log_tail, + last_fetched_at = excluded.last_fetched_at; + +-- name: GetPR :one +SELECT session_id, review_decision, mergeability, ci_state, ci_passed, ci_failed, ci_pending, ci_log_tail, last_fetched_at +FROM pr +WHERE session_id = ?; + +-- name: DeletePR :exec +DELETE FROM pr WHERE session_id = ?; + +-- name: DeletePRChecks :exec +DELETE FROM pr_check WHERE session_id = ?; + +-- name: InsertPRCheck :exec +INSERT INTO pr_check (session_id, name, status, url) VALUES (?, ?, ?, ?); + +-- name: ListPRChecks :many +SELECT name, status, url FROM pr_check WHERE session_id = ? ORDER BY name; + +-- name: DeletePRComments :exec +DELETE FROM pr_comment WHERE session_id = ?; + +-- name: InsertPRComment :exec +INSERT INTO pr_comment (session_id, comment_id, author, file, line, body, resolved, created_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?); + +-- name: ListPRComments :many +SELECT comment_id, author, file, line, body, resolved, created_at +FROM pr_comment +WHERE session_id = ? +ORDER BY created_at, comment_id; diff --git a/backend/internal/storage/sqlite/queries/pr_enrichment.sql b/backend/internal/storage/sqlite/queries/pr_enrichment.sql deleted file mode 100644 index 7c2ac0a03..000000000 --- a/backend/internal/storage/sqlite/queries/pr_enrichment.sql +++ /dev/null @@ -1,18 +0,0 @@ --- name: UpsertPREnrichment :exec -INSERT INTO pr_enrichment (session_id, ci_summary, review_decision, mergeability, pending_comments, ci_log_tail, last_fetched_at) -VALUES (?, ?, ?, ?, ?, ?, ?) -ON CONFLICT (session_id) DO UPDATE SET - ci_summary = excluded.ci_summary, - review_decision = excluded.review_decision, - mergeability = excluded.mergeability, - pending_comments = excluded.pending_comments, - ci_log_tail = excluded.ci_log_tail, - last_fetched_at = excluded.last_fetched_at; - --- name: GetPREnrichment :one -SELECT session_id, ci_summary, review_decision, mergeability, pending_comments, ci_log_tail, last_fetched_at -FROM pr_enrichment -WHERE session_id = ?; - --- name: DeletePREnrichment :exec -DELETE FROM pr_enrichment WHERE session_id = ?; diff --git a/backend/internal/storage/sqlite/store.go b/backend/internal/storage/sqlite/store.go index bd61e73b3..75b5474a2 100644 --- a/backend/internal/storage/sqlite/store.go +++ b/backend/internal/storage/sqlite/store.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "fmt" + "time" "github.com/aoagents/agent-orchestrator/backend/internal/domain" "github.com/aoagents/agent-orchestrator/backend/internal/ports" @@ -77,42 +78,41 @@ func (s *Store) ListAll(ctx context.Context) ([]domain.SessionRecord, error) { return out, nil } -// GetMetadata returns the opaque key/value metadata for a session. -func (s *Store) GetMetadata(ctx context.Context, id domain.SessionID) (map[string]string, error) { - rows, err := s.q.GetMetadata(ctx, string(id)) +// GetMetadata returns the typed metadata for a session, or the zero value if the +// session has no metadata row yet. +func (s *Store) GetMetadata(ctx context.Context, id domain.SessionID) (domain.SessionMetadata, error) { + row, err := s.q.GetSessionMetadata(ctx, string(id)) + if errors.Is(err, sql.ErrNoRows) { + return domain.SessionMetadata{}, nil + } if err != nil { - return nil, fmt.Errorf("get metadata %s: %w", id, err) + return domain.SessionMetadata{}, fmt.Errorf("get metadata %s: %w", id, err) } - if len(rows) == 0 { - return nil, nil - } - m := make(map[string]string, len(rows)) - for _, r := range rows { - m[r.Key] = r.Value - } - return m, nil + return domain.SessionMetadata{ + Branch: row.Branch, + WorkspacePath: row.WorkspacePath, + RuntimeHandleID: row.RuntimeHandleID, + RuntimeName: row.RuntimeName, + AgentSessionID: row.AgentSessionID, + Prompt: row.Prompt, + }, nil } -// PatchMetadata merges kv into the session's metadata. It is outside the -// canonical write path: no revision bump, no CDC event. -func (s *Store) PatchMetadata(ctx context.Context, id domain.SessionID, kv map[string]string) error { - if len(kv) == 0 { +// PatchMetadata merges meta into the session's metadata. It is outside the +// canonical write path: no revision bump, no CDC event. Empty fields are left +// unchanged (see UpsertSessionMetadata), so a partial patch is non-destructive. +func (s *Store) PatchMetadata(ctx context.Context, id domain.SessionID, meta domain.SessionMetadata) error { + if meta.IsZero() { return nil } - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return fmt.Errorf("begin patch metadata: %w", err) - } - defer tx.Rollback() - qtx := s.q.WithTx(tx) - for k, v := range kv { - if err := qtx.UpsertMetadata(ctx, gen.UpsertMetadataParams{ - SessionID: string(id), - Key: k, - Value: v, - }); err != nil { - return fmt.Errorf("patch metadata %s[%s]: %w", id, k, err) - } - } - return tx.Commit() + return s.q.UpsertSessionMetadata(ctx, gen.UpsertSessionMetadataParams{ + SessionID: string(id), + Branch: meta.Branch, + WorkspacePath: meta.WorkspacePath, + RuntimeHandleID: meta.RuntimeHandleID, + RuntimeName: meta.RuntimeName, + AgentSessionID: meta.AgentSessionID, + Prompt: meta.Prompt, + UpdatedAt: time.Now().UTC(), + }) } diff --git a/backend/internal/storage/sqlite/store_test.go b/backend/internal/storage/sqlite/store_test.go index 5457855da..711f8cf1e 100644 --- a/backend/internal/storage/sqlite/store_test.go +++ b/backend/internal/storage/sqlite/store_test.go @@ -156,7 +156,7 @@ func TestGetListRoundTrip(t *testing.T) { if got.ID != "a" || got.Lifecycle.Revision != 1 || got.IssueID != "issue-1" { t.Fatalf("unexpected record: %+v", got) } - if got.Metadata != nil { + if !got.Metadata.IsZero() { t.Fatalf("Get must not reconstruct metadata, got %v", got.Metadata) } @@ -176,10 +176,11 @@ func TestMetadataSideChannel(t *testing.T) { t.Fatal(err) } - if err := s.PatchMetadata(ctx, "s1", map[string]string{"branch": "feat/x", "prompt": "do it"}); err != nil { + if err := s.PatchMetadata(ctx, "s1", domain.SessionMetadata{Branch: "feat/x", Prompt: "do it"}); err != nil { t.Fatalf("patch: %v", err) } - if err := s.PatchMetadata(ctx, "s1", map[string]string{"branch": "feat/y"}); err != nil { + // A partial patch (only Branch) must not clobber the earlier Prompt. + if err := s.PatchMetadata(ctx, "s1", domain.SessionMetadata{Branch: "feat/y"}); err != nil { t.Fatalf("patch overwrite: %v", err) } @@ -187,8 +188,8 @@ func TestMetadataSideChannel(t *testing.T) { if err != nil { t.Fatal(err) } - if m["branch"] != "feat/y" || m["prompt"] != "do it" { - t.Fatalf("metadata = %v", m) + if m.Branch != "feat/y" || m.Prompt != "do it" { + t.Fatalf("metadata = %+v", m) } // Metadata writes must not bump revision (off the canonical path). lc, _, _ := s.Load(ctx, "s1") @@ -239,7 +240,7 @@ func TestLoadGetMissing(t *testing.T) { if _, ok, err := s.Get(ctx, "nope"); ok || err != nil { t.Fatalf("Get missing: ok=%v err=%v", ok, err) } - if m, err := s.GetMetadata(ctx, "nope"); err != nil || m != nil { + if m, err := s.GetMetadata(ctx, "nope"); err != nil || !m.IsZero() { t.Fatalf("GetMetadata missing: m=%v err=%v", m, err) } } diff --git a/backend/internal/storage/sqlite/upsert.go b/backend/internal/storage/sqlite/upsert.go index 40944005a..646745168 100644 --- a/backend/internal/storage/sqlite/upsert.go +++ b/backend/internal/storage/sqlite/upsert.go @@ -82,14 +82,14 @@ func casPersist(ctx context.Context, q *gen.Queries, rec domain.SessionRecord) ( } // appendOutbox writes the change_log entry and threads its seq into a fresh -// outbox row. The change_log payload is the persisted record at its new -// revision (metadata excluded — it is not on the canonical path). +// outbox row. The change_log payload is the persisted record at its new revision +// (metadata is excluded by SessionRecord's json:"-" tag — it is not on the +// canonical path). func appendOutbox(ctx context.Context, q *gen.Queries, rec domain.SessionRecord, newRevision int, eventType ports.EventType) error { now := time.Now().UTC() payload := rec payload.Lifecycle.Revision = newRevision payload.Lifecycle.Version = domain.LifecycleVersion - payload.Metadata = nil blob, err := json.Marshal(payload) if err != nil { return fmt.Errorf("marshal change_log payload %s: %w", rec.ID, err) diff --git a/backend/main_test.go b/backend/main_test.go index 1a8d60c3f..c8f325418 100644 --- a/backend/main_test.go +++ b/backend/main_test.go @@ -127,7 +127,7 @@ func TestSnapshotSourceRebuildsState(t *testing.T) { if rec.Lifecycle.Revision != 1 { t.Errorf("payload revision = %d, want 1", rec.Lifecycle.Revision) } - if rec.Metadata != nil { + if !rec.Metadata.IsZero() { t.Errorf("snapshot payload must exclude metadata, got %v", rec.Metadata) } }