feat(backend): SQLite storage layer + CDC pipeline, LCM/reaper wiring

Add the two real outbound adapters that replace the in-memory fakeStore:
internal/storage/sqlite (persistence satisfying ports.LifecycleStore) and
internal/cdc (transactional-outbox publisher, JSONL delivery, durable
consumer). Wire them into main.go alongside the Lifecycle Manager and reaper
so the write path is live end-to-end: LCM.Upsert -> store -> outbox -> JSONL
-> broadcaster.

Storage (internal/storage/sqlite):
- modernc.org/sqlite (pure Go, no CGO) for clean cross-compile; goose
  embedded migrations; sqlc-generated typed queries under gen/.
- Atomic Upsert: session row + change_log + outbox written in one tx.
- revision is an optimistic-concurrency (CAS) check: insert requires
  revision 0 and persists 1; update requires loaded revision == stored and
  bumps +1; zero rows affected returns a revision-mismatch error.
- Metadata is an opaque map in session_metadata, off the CDC path.
- Durable reaction_trackers (fixes the in-memory-only escalation budget that
  re-fired human pages on restart).

CDC (internal/cdc):
- Publisher drains the outbox to a JSONL log; size-based rotation with a
  reset marker.
- Consumer tails via byte cursor, detects rotation (os.SameFile), resyncs
  from a full-state snapshot on gaps, and tracks a durable consumer_offsets
  cursor.
- Janitor reclaims acknowledged outbox rows.
- Broadcaster is the in-process fan-out port the FE transport will subscribe
  to (WS/SSE wiring deferred).

Composition root (main.go + *_wiring.go):
- startCDC stands up publisher/consumer/janitor + broadcaster.
- startLifecycle constructs the LCM, makes escalation budgets durable via
  WithReactionStore, teaches it to enumerate sessions via WithSessionLister,
  and starts the reaper.
- Notifier, AgentMessenger, and the reaper's runtime registry are TEMPORARY
  no-op/empty stubs (lifecycle_wiring.go) with TODO markers; see the PR
  description for how to fill them in.

Tests: contract-parity, revision CAS, outbox atomicity, CDC ordering and
idempotency, rotation/resync, janitor vacuum, reaction durability across a
simulated restart, and composition-root adapters. gofmt/build/vet clean and
go test -race ./... green.
This commit is contained in:
Pritom14 2026-05-30 16:02:07 +05:30
parent 527d9c8303
commit f5bc4c7b8c
41 changed files with 3849 additions and 11 deletions

9
.gitignore vendored
View File

@ -17,6 +17,15 @@ vendor/
/backend/backend
agent-orchestrator.yaml
# Backend runtime data artifacts (SQLite store + WAL, CDC event log).
# Created at AO_DATA_DIR (outside the repo by default); ignored here so a
# data dir pointed at the tree never gets committed.
*.db
*.db-shm
*.db-wal
session-events.jsonl
session-events.jsonl.*
# Environment
.env
.env.*

143
backend/cdc_wiring.go Normal file
View File

@ -0,0 +1,143 @@
package main
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"path/filepath"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/cdc"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
// cdcConsumerName is the durable consumer_offsets key for the in-process FE
// broadcast consumer. A second transport (e.g. a cloud relay) would use its own
// key so each tracks an independent cursor.
const cdcConsumerName = "fe-broadcast"
// cdcPipeline owns the running CDC goroutines and the broadcaster the FE
// transport subscribes to. It is the durable change-delivery substrate: the
// publisher drains the outbox to JSONL, the consumer tails the log and fans out
// through the broadcaster, and the janitor reclaims acknowledged outbox rows.
type cdcPipeline struct {
Broadcaster *cdc.Broadcaster
log *cdc.Log
dones []<-chan struct{}
}
// startCDC opens the JSONL log and starts the publisher, consumer, and janitor
// against store, returning a handle whose Stop waits for the goroutines to
// drain after ctx is cancelled. The goroutines stop when ctx is cancelled.
func startCDC(ctx context.Context, store *sqlite.Store, dataDir string, logger *slog.Logger) (*cdcPipeline, error) {
log, err := cdc.OpenLog(dataDir, 0)
if err != nil {
return nil, fmt.Errorf("open cdc log: %w", err)
}
bcast := cdc.NewBroadcaster()
logPath := filepath.Join(dataDir, cdc.LogFileName)
pub := cdc.NewPublisher(outboxAdapter{store}, log, cdc.PublisherConfig{Logger: logger})
con := cdc.NewConsumer(cdcConsumerName, logPath, store, bcast, cdc.ConsumerConfig{
Snapshot: snapshotSource{store},
Logger: logger,
})
jan := cdc.NewJanitor(store, cdc.JanitorConfig{Logger: logger})
conDone, err := con.Start(ctx)
if err != nil {
log.Close()
return nil, fmt.Errorf("start cdc consumer: %w", err)
}
return &cdcPipeline{
Broadcaster: bcast,
log: log,
dones: []<-chan struct{}{pub.Start(ctx), conDone, jan.Start(ctx)},
}, nil
}
// Stop waits for every CDC goroutine to exit (the caller must have cancelled the
// ctx passed to startCDC) and closes the log file.
func (p *cdcPipeline) Stop() error {
for _, d := range p.dones {
<-d
}
return p.log.Close()
}
// outboxAdapter bridges *sqlite.Store's outbox methods to cdc.OutboxStore,
// mapping the storage-native OutboxEvent to the transport's PendingEvent. (The
// offset and vacuum contracts need no adapter — *sqlite.Store satisfies
// cdc.OffsetStore and cdc.Vacuum directly.)
type outboxAdapter struct{ store *sqlite.Store }
func (a outboxAdapter) ListUnsent(ctx context.Context, limit int) ([]cdc.PendingEvent, error) {
evs, err := a.store.ListUnsent(ctx, limit)
if err != nil {
return nil, err
}
out := make([]cdc.PendingEvent, len(evs))
for i, e := range evs {
out[i] = cdc.PendingEvent{
OutboxID: e.OutboxID,
Event: cdc.Event{
Seq: e.Seq,
SessionID: e.SessionID,
EventType: e.EventType,
Revision: e.Revision,
Payload: e.Payload,
CreatedAt: e.CreatedAt,
},
}
}
return out, nil
}
func (a outboxAdapter) MarkSent(ctx context.Context, id int64, at time.Time) error {
return a.store.MarkSent(ctx, id, at)
}
func (a outboxAdapter) MarkFailed(ctx context.Context, id int64, msg string) error {
return a.store.MarkFailed(ctx, id, msg)
}
// snapshotSource rebuilds current state from the sessions table after a
// log-rotation gap, emitting one full-state event per session. Each event
// carries the change_log high-water seq so the consumer resumes its cursor
// there; the payload mirrors the canonical change_log payload (metadata
// excluded, version stamped) so subscribers parse snapshot and live events the
// same way.
type snapshotSource struct{ store *sqlite.Store }
func (s snapshotSource) Snapshot(ctx context.Context) ([]cdc.Event, int64, error) {
recs, err := s.store.ListAll(ctx)
if err != nil {
return nil, 0, err
}
maxSeq, err := s.store.MaxChangeLogSeq(ctx)
if err != nil {
return nil, 0, err
}
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)
}
events = append(events, cdc.Event{
Seq: maxSeq,
SessionID: string(r.ID),
EventType: "session_snapshot",
Revision: int64(r.Lifecycle.Revision),
Payload: string(blob),
CreatedAt: r.UpdatedAt,
})
}
return events, maxSeq, nil
}

View File

@ -1,5 +1,25 @@
module github.com/aoagents/agent-orchestrator/backend
go 1.22
go 1.25.7
require github.com/go-chi/chi/v5 v5.1.0
require (
github.com/go-chi/chi/v5 v5.1.0
github.com/pressly/goose/v3 v3.27.1
modernc.org/sqlite v1.51.0
)
require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/mattn/go-isatty v0.0.21 // indirect
github.com/mfridman/interpolate v0.0.2 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
github.com/sethvargo/go-retry v0.3.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.43.0 // indirect
modernc.org/libc v1.72.3 // indirect
modernc.org/mathutil v1.7.1 // indirect
modernc.org/memory v1.11.0 // indirect
)

View File

@ -1,2 +1,68 @@
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw=
github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/mattn/go-isatty v0.0.21 h1:xYae+lCNBP7QuW4PUnNG61ffM4hVIfm+zUzDuSzYLGs=
github.com/mattn/go-isatty v0.0.21/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4=
github.com/mfridman/interpolate v0.0.2 h1:pnuTK7MQIxxFz1Gr+rjSIx9u7qVjf5VOoM/u6BbAxPY=
github.com/mfridman/interpolate v0.0.2/go.mod h1:p+7uk6oE07mpE/Ik1b8EckO0O4ZXiGAfshKBWLUM9Xg=
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/pressly/goose/v3 v3.27.1 h1:6uEvcprBybDmW4hcz3gYujhARhye+GoWKhEWyzD5sh4=
github.com/pressly/goose/v3 v3.27.1/go.mod h1:maruOxsPnIG2yHHyo8UqKWXYKFcH7Q76csUV7+7KYoM=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/sethvargo/go-retry v0.3.0 h1:EEt31A35QhrcRZtrYFDTBg91cqZVnFL2navjDrah2SE=
github.com/sethvargo/go-retry v0.3.0/go.mod h1:mNX17F0C/HguQMyMyJxcnU471gOZGxCLyYaFyAZraas=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
modernc.org/cc/v4 v4.28.2 h1:3tQ0lf2ADtoby2EtSP+J7IE2SHwEJdP8ioR59wx7XpY=
modernc.org/cc/v4 v4.28.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI=
modernc.org/ccgo/v4 v4.34.0 h1:yRLPFZieg532OT4rp4JFNIVcquwalMX26G95WQDqwCQ=
modernc.org/ccgo/v4 v4.34.0/go.mod h1:AS5WYMyBakQ+fhsHhtP8mWB82KTGPkNNJDGfGQCe0/A=
modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM=
modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU=
modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
modernc.org/gc/v3 v3.1.2 h1:ZtDCnhonXSZexk/AYsegNRV1lJGgaNZJuKjJSWKyEqo=
modernc.org/gc/v3 v3.1.2/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
modernc.org/libc v1.72.3 h1:ZnDF4tXn4NBXFutMMQC4vtbTFSXhhKzR73fv0beZEAU=
modernc.org/libc v1.72.3/go.mod h1:dn0dZNnnn1clLyvRxLxYExxiKRZIRENOfqQ8XEeg4Qs=
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg=
modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
modernc.org/sqlite v1.51.0 h1:aH/MMSoayAIhozZ7uJbVTT9QO/VhzBf0J9tymmmuC/U=
modernc.org/sqlite v1.51.0/go.mod h1:tcNzv5p84E0skkmJn038y+hWJbLQXQqEnQfeh5r2JLM=
modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=

View File

@ -0,0 +1,44 @@
package cdc
import "sync"
// Broadcaster is the in-process fan-out the consumer feeds. Subscribers (the
// WS/SSE transport, wired in the frontend task) register a callback; every
// consumed Event is delivered to all current subscribers. It is the single
// seam between the CDC pipeline and live delivery, so the transport can be
// built and swapped without touching the pipeline.
type Broadcaster struct {
mu sync.RWMutex
nextID int
subs map[int]func(Event)
}
// NewBroadcaster returns an empty Broadcaster ready for subscriptions.
func NewBroadcaster() *Broadcaster {
return &Broadcaster{subs: map[int]func(Event){}}
}
// Subscribe registers fn and returns an unsubscribe function. fn is called
// synchronously from the consumer loop, so it must not block; a transport that
// needs buffering should push onto its own channel inside fn.
func (b *Broadcaster) Subscribe(fn func(Event)) (unsubscribe func()) {
b.mu.Lock()
id := b.nextID
b.nextID++
b.subs[id] = fn
b.mu.Unlock()
return func() {
b.mu.Lock()
delete(b.subs, id)
b.mu.Unlock()
}
}
// Publish delivers e to every current subscriber.
func (b *Broadcaster) Publish(e Event) {
b.mu.RLock()
defer b.mu.RUnlock()
for _, fn := range b.subs {
fn(e)
}
}

View File

@ -0,0 +1,256 @@
package cdc_test
import (
"context"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/cdc"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
// outboxAdapter bridges sqlite.Store's outbox methods to cdc.OutboxStore. This
// is the same glue the composition root (main.go) installs.
type outboxAdapter struct{ s *sqlite.Store }
func (a outboxAdapter) ListUnsent(ctx context.Context, limit int) ([]cdc.PendingEvent, error) {
evs, err := a.s.ListUnsent(ctx, limit)
if err != nil {
return nil, err
}
out := make([]cdc.PendingEvent, len(evs))
for i, e := range evs {
out[i] = cdc.PendingEvent{
OutboxID: e.OutboxID,
Event: cdc.Event{
Seq: e.Seq,
SessionID: e.SessionID,
EventType: e.EventType,
Revision: e.Revision,
Payload: e.Payload,
CreatedAt: e.CreatedAt,
},
}
}
return out, nil
}
func (a outboxAdapter) MarkSent(ctx context.Context, id int64, at time.Time) error {
return a.s.MarkSent(ctx, id, at)
}
func (a outboxAdapter) MarkFailed(ctx context.Context, id int64, msg string) error {
return a.s.MarkFailed(ctx, id, msg)
}
func newStore(t *testing.T) *sqlite.Store {
t.Helper()
db, err := sqlite.Open(t.TempDir())
if err != nil {
t.Fatalf("open: %v", err)
}
t.Cleanup(func() { db.Close() })
return sqlite.NewStore(db)
}
func rec(id string) domain.SessionRecord {
now := time.Now().UTC()
return domain.SessionRecord{
ID: domain.SessionID(id), ProjectID: "p", Kind: domain.KindWorker, CreatedAt: now, UpdatedAt: now,
Lifecycle: domain.CanonicalSessionLifecycle{
Session: domain.SessionSubstate{State: domain.SessionWorking, Reason: domain.ReasonTaskInProgress},
PR: domain.PRSubstate{State: domain.PRNone, Reason: domain.PRReasonNotCreated},
Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning},
Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: now, Source: domain.SourceNative},
},
}
}
func TestEndToEndPublishConsume(t *testing.T) {
ctx := context.Background()
store := newStore(t)
dir := t.TempDir()
log, err := cdc.OpenLog(dir, 0)
if err != nil {
t.Fatal(err)
}
defer log.Close()
// Three canonical writes => three outbox rows, seq 1..3.
r := rec("s1")
if err := store.Upsert(ctx, r, ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
r.Lifecycle.Revision = 1
if err := store.Upsert(ctx, r, ports.EventSessionStateChanged); err != nil {
t.Fatal(err)
}
r.Lifecycle.Revision = 2
if err := store.Upsert(ctx, r, ports.EventSessionStateChanged); err != nil {
t.Fatal(err)
}
pub := cdc.NewPublisher(outboxAdapter{store}, log, cdc.PublisherConfig{})
if err := pub.Drain(ctx); err != nil {
t.Fatalf("drain: %v", err)
}
var got []cdc.Event
bc := cdc.NewBroadcaster()
bc.Subscribe(func(e cdc.Event) { got = append(got, e) })
con := cdc.NewConsumer("fe", dir+"/"+cdc.LogFileName, store, bc, cdc.ConsumerConfig{})
if _, err := con.Start(ctx); err != nil {
t.Fatal(err)
}
// Drive one poll synchronously instead of waiting on the goroutine.
if err := con.Poll(ctx); err != nil {
t.Fatalf("poll: %v", err)
}
if len(got) != 3 {
t.Fatalf("delivered %d events, want 3", len(got))
}
for i, e := range got {
if e.Seq != int64(i+1) {
t.Fatalf("event %d has seq %d, want %d", i, e.Seq, i+1)
}
}
if got[0].EventType != string(ports.EventSessionCreated) {
t.Fatalf("first event type = %q", got[0].EventType)
}
// Idempotency: a second poll with no new bytes delivers nothing more.
if err := con.Poll(ctx); err != nil {
t.Fatal(err)
}
if len(got) != 3 {
t.Fatalf("re-poll delivered extra events: %d", len(got))
}
// Offset persisted at seq 3.
off, _ := store.GetOffset(ctx, "fe")
if off != 3 {
t.Fatalf("offset = %d, want 3", off)
}
// Janitor: consumer ACKed 3, so sent rows with seq < 3 are reclaimed.
jan := cdc.NewJanitor(store, cdc.JanitorConfig{})
deleted, err := jan.Sweep(ctx)
if err != nil {
t.Fatal(err)
}
if deleted != 2 {
t.Fatalf("janitor deleted %d, want 2 (seq 1,2 < watermark 3)", deleted)
}
}
func TestConsumerRestartSkipsDelivered(t *testing.T) {
ctx := context.Background()
store := newStore(t)
dir := t.TempDir()
log, _ := cdc.OpenLog(dir, 0)
defer log.Close()
if err := store.Upsert(ctx, rec("s1"), ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
pub := cdc.NewPublisher(outboxAdapter{store}, log, cdc.PublisherConfig{})
if err := pub.Drain(ctx); err != nil {
t.Fatal(err)
}
// Pre-seed the durable offset as if a prior consumer already delivered seq 1.
if err := store.SetOffset(ctx, "fe", 1, time.Now().UTC()); err != nil {
t.Fatal(err)
}
var got []cdc.Event
bc := cdc.NewBroadcaster()
bc.Subscribe(func(e cdc.Event) { got = append(got, e) })
con := cdc.NewConsumer("fe", dir+"/"+cdc.LogFileName, store, bc, cdc.ConsumerConfig{})
if _, err := con.Start(ctx); err != nil {
t.Fatal(err)
}
if err := con.Poll(ctx); err != nil {
t.Fatal(err)
}
if len(got) != 0 {
t.Fatalf("restart re-delivered already-acked events: %d", len(got))
}
}
// fakeSnapshot stands in for the sessions-table snapshot source on resync.
type fakeSnapshot struct {
events []cdc.Event
maxSeq int64
}
func (f fakeSnapshot) Snapshot(context.Context) ([]cdc.Event, int64, error) {
return f.events, f.maxSeq, nil
}
func TestRotationTriggersResync(t *testing.T) {
ctx := context.Background()
store := newStore(t)
dir := t.TempDir()
// Tiny cap so a couple of writes force a rotation.
log, err := cdc.OpenLog(dir, 80)
if err != nil {
t.Fatal(err)
}
defer log.Close()
var got []cdc.Event
bc := cdc.NewBroadcaster()
bc.Subscribe(func(e cdc.Event) { got = append(got, e) })
snap := fakeSnapshot{events: []cdc.Event{{Seq: 5, SessionID: "s1", EventType: "session_updated"}}, maxSeq: 5}
con := cdc.NewConsumer("fe", dir+"/"+cdc.LogFileName, store, bc, cdc.ConsumerConfig{Snapshot: snap})
if _, err := con.Start(ctx); err != nil {
t.Fatal(err)
}
pub := cdc.NewPublisher(outboxAdapter{store}, log, cdc.PublisherConfig{})
// First write + drain + poll: consumer reads it and advances its cursor.
if err := store.Upsert(ctx, rec("s1"), ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
if err := pub.Drain(ctx); err != nil {
t.Fatal(err)
}
if err := con.Poll(ctx); err != nil {
t.Fatal(err)
}
cursorBefore := len(got)
// Force rotation by writing past the cap, then poll: the file shrank, so the
// consumer must resync from the snapshot source.
r := rec("s1")
r.Lifecycle.Revision = 1
if err := store.Upsert(ctx, r, ports.EventSessionStateChanged); err != nil {
t.Fatal(err)
}
if err := pub.Drain(ctx); err != nil {
t.Fatal(err)
}
if err := con.Poll(ctx); err != nil {
t.Fatal(err)
}
if len(got) <= cursorBefore {
t.Fatal("expected resync to deliver the snapshot event")
}
// The snapshot event (seq 5) must be among the delivered events.
var sawSnapshot bool
for _, e := range got {
if e.Seq == 5 {
sawSnapshot = true
}
}
if !sawSnapshot {
t.Fatalf("resync did not deliver snapshot event; got %+v", got)
}
}

View File

@ -0,0 +1,221 @@
package cdc
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"os"
"time"
)
// DefaultPollInterval is how often the consumer checks the log for new bytes.
// Polling (rather than fs-notify) keeps the consumer dependency-free; at this
// cadence live updates stay well under a human-perceptible delay.
const DefaultPollInterval = 100 * time.Millisecond
// OffsetStore persists the consumer's durable seq cursor (at-least-once).
type OffsetStore interface {
GetOffset(ctx context.Context, consumer string) (int64, error)
SetOffset(ctx context.Context, consumer string, seq int64, at time.Time) error
}
// SnapshotSource rebuilds current state from the source of truth (the sessions
// table) after a rotation gap, where log lines for unconsumed-but-already-sent
// events were truncated away. It returns one Event per live session plus the
// MAX(change_log seq) the snapshot corresponds to, so the consumer can resume.
type SnapshotSource interface {
Snapshot(ctx context.Context) (events []Event, maxSeq int64, err error)
}
// Consumer tails the JSONL log, deduplicates by seq, and fans each new event
// out through the Broadcaster, persisting its durable offset as it goes.
type Consumer struct {
name string
path string
offsets OffsetStore
bcast *Broadcaster
snapshot SnapshotSource
interval time.Duration
clock func() time.Time
logger *slog.Logger
cursor int64 // byte offset into the log
lastSeq int64 // highest seq delivered
prevInfo os.FileInfo // identity of the file last polled (rotation detection)
}
// ConsumerConfig holds optional knobs and the snapshot source.
type ConsumerConfig struct {
Snapshot SnapshotSource
Interval time.Duration
Clock func() time.Time
Logger *slog.Logger
}
// NewConsumer constructs a Consumer named name (the consumer_offsets key) over
// the log at path, fanning out through bcast and persisting offsets via offsets.
func NewConsumer(name, path string, offsets OffsetStore, bcast *Broadcaster, cfg ConsumerConfig) *Consumer {
c := &Consumer{
name: name,
path: path,
offsets: offsets,
bcast: bcast,
snapshot: cfg.Snapshot,
interval: cfg.Interval,
clock: cfg.Clock,
logger: cfg.Logger,
}
if c.interval <= 0 {
c.interval = DefaultPollInterval
}
if c.clock == nil {
c.clock = time.Now
}
if c.logger == nil {
c.logger = slog.Default()
}
return c
}
// Start loads the durable offset and runs the poll loop until ctx is cancelled;
// the returned channel closes when the loop has exited.
func (c *Consumer) Start(ctx context.Context) (<-chan struct{}, error) {
seq, err := c.offsets.GetOffset(ctx, c.name)
if err != nil {
return nil, fmt.Errorf("load consumer offset: %w", err)
}
c.lastSeq = seq
done := make(chan struct{})
go func() {
defer close(done)
t := time.NewTicker(c.interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
if err := c.Poll(ctx); err != nil {
c.logger.Error("cdc consumer: poll failed", "err", err)
}
}
}
}()
return done, nil
}
// Poll reads any new bytes since the last cursor and delivers complete lines. It
// detects rotation (the file shrank below the cursor) and resyncs from the DB
// snapshot before resuming.
func (c *Consumer) Poll(ctx context.Context) error {
f, err := os.Open(c.path)
if err != nil {
if os.IsNotExist(err) {
return nil // publisher has not created the log yet
}
return fmt.Errorf("open cdc log: %w", err)
}
defer f.Close()
info, err := f.Stat()
if err != nil {
return fmt.Errorf("stat cdc log: %w", err)
}
size := info.Size()
rotated := (c.prevInfo != nil && !os.SameFile(c.prevInfo, info)) || size < c.cursor
c.prevInfo = info
if rotated {
// The previous file's bytes are void. Resync from the DB snapshot (if
// wired), then resume reading the fresh file from the top.
if err := c.resync(ctx); err != nil {
return err
}
c.cursor = 0
}
if size == c.cursor {
return nil
}
if _, err := f.Seek(c.cursor, io.SeekStart); err != nil {
return fmt.Errorf("seek cdc log: %w", err)
}
data, err := io.ReadAll(f)
if err != nil {
return fmt.Errorf("read cdc log: %w", err)
}
consumed, maxSeq := c.processLines(data)
c.cursor += int64(consumed)
if maxSeq > c.lastSeq {
c.lastSeq = maxSeq
if err := c.offsets.SetOffset(ctx, c.name, c.lastSeq, c.clock().UTC()); err != nil {
return fmt.Errorf("persist consumer offset: %w", err)
}
}
return nil
}
// processLines delivers each complete (newline-terminated) line, skipping reset
// markers and any event whose seq was already delivered. It returns the number
// of bytes consumed (only complete lines) and the highest seq seen.
func (c *Consumer) processLines(data []byte) (consumed int, maxSeq int64) {
maxSeq = c.lastSeq
for {
nl := bytes.IndexByte(data[consumed:], '\n')
if nl < 0 {
return consumed, maxSeq // partial trailing line: leave for next poll
}
line := data[consumed : consumed+nl]
consumed += nl + 1
if isResetMarker(line) {
continue
}
var e Event
if err := json.Unmarshal(line, &e); err != nil {
c.logger.Error("cdc consumer: bad line skipped", "err", err)
continue
}
if e.Seq <= c.lastSeq {
continue // idempotent: already delivered
}
c.bcast.Publish(e)
if e.Seq > maxSeq {
maxSeq = e.Seq
}
}
}
func (c *Consumer) resync(ctx context.Context) error {
if c.snapshot == nil {
return nil
}
events, maxSeq, err := c.snapshot.Snapshot(ctx)
if err != nil {
return fmt.Errorf("cdc consumer resync: %w", err)
}
for _, e := range events {
c.bcast.Publish(e)
}
if maxSeq > c.lastSeq {
c.lastSeq = maxSeq
if err := c.offsets.SetOffset(ctx, c.name, c.lastSeq, c.clock().UTC()); err != nil {
return fmt.Errorf("persist offset after resync: %w", err)
}
}
return nil
}
func isResetMarker(line []byte) bool {
var m resetMarker
if err := json.Unmarshal(line, &m); err != nil {
return false
}
return m.Type == "reset"
}

View File

@ -0,0 +1,32 @@
// Package cdc is the change-data-capture pipeline that turns the storage layer's
// transactional outbox into a durable, ordered event stream for the frontend.
//
// The flow: the publisher drains the SQLite outbox (sent=0, seq order) and
// appends each change as one JSON line to a rotating log file. The consumer
// tails that file from a durable byte cursor, deduplicates by seq, and fans each
// change out through the Broadcaster to in-process subscribers (the WS/SSE
// transport, wired later). The janitor reclaims outbox rows every consumer has
// acknowledged. Delivery is at-least-once; seq is the idempotency key.
package cdc
import "time"
// Event is one change-data-capture record. It is the JSONL line shape and the
// value handed to Broadcaster subscribers. Seq is the monotonic ordering and
// idempotency key (the change_log seq).
type Event struct {
Seq int64 `json:"seq"`
SessionID string `json:"sessionId"`
EventType string `json:"eventType"`
Revision int64 `json:"revision"`
Payload string `json:"payload"`
CreatedAt time.Time `json:"createdAt"`
}
// resetMarker is written as the first line of a freshly rotated log file. A
// consumer that reads it knows the byte offsets of the previous file are void
// and must snapshot-resync, then resume from the current MAX(seq).
type resetMarker struct {
Type string `json:"type"` // always "reset"
RotatedAt time.Time `json:"rotatedAt"`
}

View File

@ -0,0 +1,84 @@
package cdc
import (
"context"
"log/slog"
"time"
)
// DefaultJanitorInterval is the outbox-vacuum cadence.
const DefaultJanitorInterval = 60 * time.Second
// Vacuum is the janitor's view of storage: the safe deletion watermark and the
// delete itself.
type Vacuum interface {
MinConsumerOffset(ctx context.Context) (int64, error)
DeleteSentOutboxBelow(ctx context.Context, seq int64) (int64, error)
}
// Janitor reclaims delivered outbox rows every consumer has acknowledged.
//
// Watermark: MIN(consumer_offsets.last_seq). Rows with seq < watermark are sent
// AND past every consumer's cursor, so they are safe to drop. When the watermark
// is 0 (a consumer exists but has acknowledged nothing, or none is registered
// yet) the janitor deletes nothing — it never races ahead of a consumer that
// has not yet read an event. change_log is never touched: it is the durable
// history and the snapshot-resync floor.
type Janitor struct {
store Vacuum
interval time.Duration
logger *slog.Logger
}
// JanitorConfig holds optional knobs; zero values fall back to defaults.
type JanitorConfig struct {
Interval time.Duration
Logger *slog.Logger
}
// NewJanitor constructs a Janitor over store.
func NewJanitor(store Vacuum, cfg JanitorConfig) *Janitor {
j := &Janitor{store: store, interval: cfg.Interval, logger: cfg.Logger}
if j.interval <= 0 {
j.interval = DefaultJanitorInterval
}
if j.logger == nil {
j.logger = slog.Default()
}
return j
}
// Start runs the vacuum loop until ctx is cancelled; the returned channel closes
// when the loop has exited.
func (j *Janitor) Start(ctx context.Context) <-chan struct{} {
done := make(chan struct{})
go func() {
defer close(done)
t := time.NewTicker(j.interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
if _, err := j.Sweep(ctx); err != nil {
j.logger.Error("cdc janitor: sweep failed", "err", err)
}
}
}
}()
return done
}
// Sweep deletes delivered outbox rows below the safe watermark and returns the
// number removed.
func (j *Janitor) Sweep(ctx context.Context) (int64, error) {
watermark, err := j.store.MinConsumerOffset(ctx)
if err != nil {
return 0, err
}
if watermark <= 0 {
return 0, nil
}
return j.store.DeleteSentOutboxBelow(ctx, watermark)
}

View File

@ -0,0 +1,109 @@
package cdc
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"sync"
"time"
)
// LogFileName is the active CDC log under the data dir.
const LogFileName = "session-events.jsonl"
// DefaultMaxBytes is the size at which the log rotates (1 MiB).
const DefaultMaxBytes int64 = 1 << 20
// Log is the append-only JSONL sink the publisher writes to. When it grows past
// maxBytes it rotates by truncating in place and writing a reset marker as the
// new first line — the consumer treats a shrunken file as "resync from the DB
// snapshot", so the log itself is not the durable source of truth (SQLite is).
type Log struct {
mu sync.Mutex
path string
maxBytes int64
f *os.File
size int64
}
// OpenLog opens (creating if absent) the JSONL log in dir. maxBytes <= 0 uses
// DefaultMaxBytes.
func OpenLog(dir string, maxBytes int64) (*Log, error) {
if maxBytes <= 0 {
maxBytes = DefaultMaxBytes
}
path := filepath.Join(dir, LogFileName)
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return nil, fmt.Errorf("open cdc log: %w", err)
}
info, err := f.Stat()
if err != nil {
f.Close()
return nil, fmt.Errorf("stat cdc log: %w", err)
}
return &Log{path: path, maxBytes: maxBytes, f: f, size: info.Size()}, nil
}
// Append writes one event as a JSON line, flushing to disk. It rotates first if
// the file is already at/over the size cap, so a single oversized burst still
// lands in a fresh segment.
func (l *Log) Append(e Event) error {
l.mu.Lock()
defer l.mu.Unlock()
if l.size >= l.maxBytes {
if err := l.rotateLocked(); err != nil {
return err
}
}
return l.writeLocked(e)
}
func (l *Log) writeLocked(v any) error {
line, err := json.Marshal(v)
if err != nil {
return fmt.Errorf("marshal cdc line: %w", err)
}
line = append(line, '\n')
n, err := l.f.Write(line)
l.size += int64(n)
if err != nil {
return fmt.Errorf("write cdc line: %w", err)
}
if err := l.f.Sync(); err != nil {
return fmt.Errorf("sync cdc log: %w", err)
}
return nil
}
// rotateLocked renames the active file aside and starts a fresh one whose first
// line is a reset marker. Renaming (not truncating in place) gives the file a
// new identity, so a polling consumer reliably detects rotation via
// os.SameFile even if the fresh file grows past its old byte cursor between
// polls. The consumer then resyncs from the DB snapshot.
func (l *Log) rotateLocked() error {
if err := l.f.Close(); err != nil {
return fmt.Errorf("close cdc log for rotate: %w", err)
}
archive := l.path + ".1"
_ = os.Remove(archive) // best-effort: history lives in SQLite, not the log
if err := os.Rename(l.path, archive); err != nil {
return fmt.Errorf("rotate cdc log: %w", err)
}
f, err := os.OpenFile(l.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return fmt.Errorf("reopen cdc log after rotate: %w", err)
}
l.f = f
l.size = 0
return l.writeLocked(resetMarker{Type: "reset", RotatedAt: time.Now().UTC()})
}
// Close closes the underlying file.
func (l *Log) Close() error {
l.mu.Lock()
defer l.mu.Unlock()
return l.f.Close()
}

View File

@ -0,0 +1,115 @@
package cdc
import (
"context"
"log/slog"
"time"
)
// DefaultPublishInterval is the outbox drain cadence.
const DefaultPublishInterval = 50 * time.Millisecond
// DefaultBatchSize bounds how many outbox rows one drain pass handles.
const DefaultBatchSize = 256
// PendingEvent is an undelivered outbox row paired with its CDC event payload.
type PendingEvent struct {
OutboxID int64
Event
}
// OutboxStore is the publisher's view of the storage layer: read undelivered
// rows in seq order, then mark each delivered or failed.
type OutboxStore interface {
ListUnsent(ctx context.Context, limit int) ([]PendingEvent, error)
MarkSent(ctx context.Context, outboxID int64, at time.Time) error
MarkFailed(ctx context.Context, outboxID int64, errMsg string) error
}
// Publisher drains the outbox into the JSONL log on a fixed cadence.
type Publisher struct {
src OutboxStore
log *Log
interval time.Duration
batch int
clock func() time.Time
logger *slog.Logger
}
// PublisherConfig holds optional knobs; zero values fall back to defaults.
type PublisherConfig struct {
Interval time.Duration
Batch int
Clock func() time.Time
Logger *slog.Logger
}
// NewPublisher constructs a Publisher over src and log.
func NewPublisher(src OutboxStore, log *Log, cfg PublisherConfig) *Publisher {
p := &Publisher{
src: src,
log: log,
interval: cfg.Interval,
batch: cfg.Batch,
clock: cfg.Clock,
logger: cfg.Logger,
}
if p.interval <= 0 {
p.interval = DefaultPublishInterval
}
if p.batch <= 0 {
p.batch = DefaultBatchSize
}
if p.clock == nil {
p.clock = time.Now
}
if p.logger == nil {
p.logger = slog.Default()
}
return p
}
// Start runs the drain loop until ctx is cancelled; the returned channel closes
// when the loop has exited.
func (p *Publisher) Start(ctx context.Context) <-chan struct{} {
done := make(chan struct{})
go func() {
defer close(done)
t := time.NewTicker(p.interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
if err := p.Drain(ctx); err != nil {
p.logger.Error("cdc publisher: drain failed", "err", err)
}
}
}
}()
return done
}
// Drain runs one pass: append each undelivered row to the log in seq order,
// marking it sent. A write failure stops the pass (the row is marked failed and
// retried next tick) so ordering is never violated by skipping ahead.
func (p *Publisher) Drain(ctx context.Context) error {
pending, err := p.src.ListUnsent(ctx, p.batch)
if err != nil {
return err
}
for _, pe := range pending {
if err := p.log.Append(pe.Event); err != nil {
p.logger.Error("cdc publisher: append failed", "outboxId", pe.OutboxID, "seq", pe.Seq, "err", err)
if merr := p.src.MarkFailed(ctx, pe.OutboxID, err.Error()); merr != nil {
p.logger.Error("cdc publisher: mark failed errored", "outboxId", pe.OutboxID, "err", merr)
}
return nil
}
if err := p.src.MarkSent(ctx, pe.OutboxID, p.clock().UTC()); err != nil {
return err
}
}
return nil
}

View File

@ -47,6 +47,9 @@ type Config struct {
// RunFilePath is where the PID + port handshake file (running.json) is
// written so the Electron supervisor can discover and reap the daemon.
RunFilePath string
// DataDir is the directory holding durable state (the SQLite database and
// the CDC JSONL log). It is created on first use by the storage layer.
DataDir string
}
// Addr returns the host:port the HTTP server binds. It uses net.JoinHostPort so
@ -65,6 +68,7 @@ func (c Config) Addr() string {
// AO_REQUEST_TIMEOUT per-request timeout (Go duration > 0, default 60s)
// AO_SHUTDOWN_TIMEOUT shutdown deadline (Go duration > 0, default 10s)
// AO_RUN_FILE running.json path (default <state-dir>/running.json)
// AO_DATA_DIR durable state dir (default <state-dir>/data)
//
// The bind host is not configurable: the daemon is loopback-only by design.
func Load() (Config, error) {
@ -108,6 +112,12 @@ func Load() (Config, error) {
}
cfg.RunFilePath = runFile
dataDir, err := resolveDataDir()
if err != nil {
return Config{}, err
}
cfg.DataDir = dataDir
return cfg, nil
}
@ -138,3 +148,17 @@ func resolveRunFilePath() (string, error) {
}
return filepath.Join(dir, "agent-orchestrator", "running.json"), nil
}
// resolveDataDir picks where durable state (SQLite DB, CDC JSONL) lives. An
// explicit AO_DATA_DIR wins; otherwise it sits under the per-user state
// directory alongside running.json.
func resolveDataDir() (string, error) {
if p, ok := os.LookupEnv("AO_DATA_DIR"); ok && p != "" {
return p, nil
}
dir, err := os.UserConfigDir()
if err != nil {
return "", fmt.Errorf("resolve state dir: %w", err)
}
return filepath.Join(dir, "agent-orchestrator", "data"), nil
}

View File

@ -53,6 +53,11 @@ type Manager struct {
trackerMu sync.Mutex
clock func() time.Time
// reactionStore, when wired via WithReactionStore, makes the trackers map a
// write-through cache over durable rows so a restart does not re-fire an
// already-escalated human page. nil keeps the in-memory-only default.
reactionStore ReactionStore
// sessionLister returns every session known to persistence so RunningSessions
// can filter by runtime axis without coupling the LCM to a cross-project
// store API the Tom-store does not yet expose. The daemon (lane #10) injects
@ -423,7 +428,7 @@ func (m *Manager) OnKillRequested(ctx context.Context, id domain.SessionID, r po
// A kill is terminal but bypasses react()'s incident-over cleanup (it fires
// no reaction). Drop any escalation trackers here so a later duration-based
// TickEscalations can't emit reaction.escalated for a dead session.
m.clearSessionTrackers(id)
m.clearSessionTrackers(ctx, id)
return nil
}

View File

@ -0,0 +1,144 @@
package lifecycle
import (
"context"
"testing"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
// TestStoreParity is the key contract test from the plan: it drives the REAL
// Lifecycle Manager through identical operation sequences against the in-memory
// fakeStore (the authoritative store semantics) and the SQLite-backed Store,
// then asserts the resulting canonical lifecycle is byte-identical. If the
// SQLite adapter honored the port exactly, the two managers cannot diverge.
//
// Both stores are seeded the same way (via the public Upsert insert path, so
// both start at revision 1) — this makes revision numbers, not just states,
// directly comparable.
func TestStoreParity(t *testing.T) {
seed := lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive)
seed.Activity = domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: t0, Source: domain.SourceNative}
cases := []struct {
name string
ops []func(*Manager) error
}{
{
name: "runtime dead then activity signal",
ops: []func(*Manager) error{
func(m *Manager) error {
return m.ApplyRuntimeObservation(context.Background(), sid, ports.RuntimeFacts{
RuntimeState: ports.RuntimeProbeDead, ProcessState: ports.ProcessProbeDead, ObservedAt: t0,
})
},
func(m *Manager) error {
return m.ApplyActivitySignal(context.Background(), sid, ports.ActivitySignal{
State: ports.SignalValid, Activity: domain.ActivityActive, Timestamp: t0, Source: domain.SourceHook,
})
},
},
},
{
name: "scm pr open then changes requested",
ops: []func(*Manager) error{
func(m *Manager) error {
return m.ApplySCMObservation(context.Background(), sid, ports.SCMFacts{
Fetched: true, PRState: domain.PROpen, PRNumber: 7, PRURL: "http://x/7",
})
},
},
},
{
name: "kill request terminates",
ops: []func(*Manager) error{
func(m *Manager) error {
return m.OnKillRequested(context.Background(), sid, ports.KillReason{Kind: ports.KillManual, Detail: "x"})
},
},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
fakeMgr, fakeS := newManager()
sqlMgr, sqlS := newSQLiteManager(t)
seedViaUpsert(t, fakeS, seed)
seedViaUpsert(t, sqlS, seed)
for i, op := range tc.ops {
errF := op(fakeMgr)
errS := op(sqlMgr)
if (errF == nil) != (errS == nil) {
t.Fatalf("op %d error divergence: fake=%v sqlite=%v", i, errF, errS)
}
}
fl, okF, _ := fakeS.Load(context.Background(), sid)
sl, okS, _ := sqlS.Load(context.Background(), sid)
if okF != okS {
t.Fatalf("presence divergence: fake=%v sqlite=%v", okF, okS)
}
assertLifecycleEqual(t, fl, sl)
})
}
}
func newSQLiteManager(t *testing.T) (*Manager, *sqlite.Store) {
t.Helper()
db, err := sqlite.Open(t.TempDir())
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
t.Cleanup(func() { db.Close() })
store := sqlite.NewStore(db)
return New(store, &recordingNotifier{}, &recordingMessenger{}), store
}
func seedViaUpsert(t *testing.T, store ports.LifecycleStore, l domain.CanonicalSessionLifecycle) {
t.Helper()
rec := domain.SessionRecord{
ID: sid,
ProjectID: "proj",
Kind: domain.KindWorker,
CreatedAt: t0,
UpdatedAt: t0,
Lifecycle: l,
}
if err := store.Upsert(context.Background(), rec, ports.EventSessionCreated); err != nil {
t.Fatalf("seed upsert: %v", err)
}
}
func assertLifecycleEqual(t *testing.T, a, b domain.CanonicalSessionLifecycle) {
t.Helper()
if a.Revision != b.Revision {
t.Errorf("revision: fake=%d sqlite=%d", a.Revision, b.Revision)
}
if a.Session != b.Session {
t.Errorf("session: fake=%+v sqlite=%+v", a.Session, b.Session)
}
if a.PR != b.PR {
t.Errorf("pr: fake=%+v sqlite=%+v", a.PR, b.PR)
}
if a.Runtime != b.Runtime {
t.Errorf("runtime: fake=%+v sqlite=%+v", a.Runtime, b.Runtime)
}
if a.Activity.State != b.Activity.State || a.Activity.Source != b.Activity.Source ||
!a.Activity.LastActivityAt.Equal(b.Activity.LastActivityAt) {
t.Errorf("activity: fake=%+v sqlite=%+v", a.Activity, b.Activity)
}
switch {
case a.Detecting == nil && b.Detecting == nil:
case a.Detecting == nil || b.Detecting == nil:
t.Errorf("detecting presence: fake=%v sqlite=%v", a.Detecting, b.Detecting)
default:
if a.Detecting.Attempts != b.Detecting.Attempts || a.Detecting.EvidenceHash != b.Detecting.EvidenceHash ||
!a.Detecting.StartedAt.Equal(b.Detecting.StartedAt) {
t.Errorf("detecting: fake=%+v sqlite=%+v", a.Detecting, b.Detecting)
}
}
}

View File

@ -0,0 +1,140 @@
package lifecycle
import (
"context"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
// reactionStoreAdapter bridges the concrete *sqlite.Store to the lifecycle
// package's ReactionStore interface (string/row types <-> domain types). This is
// the same glue the composition root installs.
type reactionStoreAdapter struct{ s *sqlite.Store }
func (a reactionStoreAdapter) LoadReactionTrackers(ctx context.Context) ([]PersistedTracker, error) {
rows, err := a.s.ListReactionTrackers(ctx)
if err != nil {
return nil, err
}
out := make([]PersistedTracker, len(rows))
for i, r := range rows {
out[i] = PersistedTracker{
SessionID: domain.SessionID(r.SessionID),
Key: r.ReactionKey,
Attempts: r.Attempts,
Escalated: r.Escalated,
FirstAttemptAt: r.FirstAttemptAt,
ProjectID: domain.ProjectID(r.ProjectID),
}
}
return out, nil
}
func (a reactionStoreAdapter) SaveReactionTracker(ctx context.Context, t PersistedTracker) error {
return a.s.SaveReactionTracker(ctx, sqlite.ReactionTrackerRow{
SessionID: string(t.SessionID),
ReactionKey: t.Key,
Attempts: t.Attempts,
Escalated: t.Escalated,
FirstAttemptAt: t.FirstAttemptAt,
ProjectID: string(t.ProjectID),
})
}
func (a reactionStoreAdapter) DeleteReactionTracker(ctx context.Context, id domain.SessionID, key string) error {
return a.s.DeleteReactionTracker(ctx, string(id), key)
}
func (a reactionStoreAdapter) DeleteSessionReactionTrackers(ctx context.Context, id domain.SessionID) error {
return a.s.DeleteSessionReactionTrackers(ctx, string(id))
}
// TestReaction_DurabilitySurvivesRestart is the plan's reaction_trackers
// durability check: once a reaction has escalated, a daemon restart (a fresh
// Manager hydrated from the same store) must NOT re-fire the human page — the
// exact failure the in-memory-only version had.
func TestReaction_DurabilitySurvivesRestart(t *testing.T) {
db, err := sqlite.Open(t.TempDir())
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
t.Cleanup(func() { db.Close() })
store := sqlite.NewStore(db)
adapter := reactionStoreAdapter{store}
// --- first process lifetime: drive ci-failed to escalation ---
notf1 := &recordingNotifier{}
m1 := New(store, notf1, &recordingMessenger{})
m1.clock = func() time.Time { return t0 }
if err := m1.WithReactionStore(context.Background(), adapter); err != nil {
t.Fatalf("hydrate m1: %v", err)
}
seedViaUpsert(t, store, lcOpenPR(domain.PRReasonReviewPending))
// ci-failed: retries 2, persistent → escalate on the third failure.
for i := 0; i < 4; i++ {
failCI(t, m1)
pendingCI(t, m1)
}
if c := notifyCount(notf1, "reaction.escalated"); c != 1 {
t.Fatalf("precondition: want one escalation in first lifetime, got %d", c)
}
// --- simulated restart: a fresh Manager hydrated from the same store ---
notf2 := &recordingNotifier{}
msgr2 := &recordingMessenger{}
m2 := New(store, notf2, msgr2)
m2.clock = func() time.Time { return t0 }
if err := m2.WithReactionStore(context.Background(), adapter); err != nil {
t.Fatalf("hydrate m2: %v", err)
}
// The ci-failed tracker rehydrates with escalated=true, so further failures
// are silenced: no new send-to-agent, no re-escalation.
failCI(t, m2)
if c := notifyCount(notf2, "reaction.escalated"); c != 0 {
t.Errorf("restart re-fired an already-escalated page: got %d escalations", c)
}
if len(msgr2.sent) != 0 {
t.Errorf("restart re-sent to agent despite escalated budget: got %d sends", len(msgr2.sent))
}
}
// TestReaction_DurabilityClearsOnIncidentOver proves the durable rows are
// removed when an incident resolves, so a later unrelated incident starts from a
// fresh budget rather than a stale escalated=true.
func TestReaction_DurabilityClearsOnIncidentOver(t *testing.T) {
db, err := sqlite.Open(t.TempDir())
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
t.Cleanup(func() { db.Close() })
store := sqlite.NewStore(db)
adapter := reactionStoreAdapter{store}
m := New(store, &recordingNotifier{}, &recordingMessenger{})
m.clock = func() time.Time { return t0 }
if err := m.WithReactionStore(context.Background(), adapter); err != nil {
t.Fatalf("hydrate: %v", err)
}
seedViaUpsert(t, store, lcOpenPR(domain.PRReasonReviewPending))
failCI(t, m)
if rows, _ := store.ListReactionTrackers(context.Background()); len(rows) == 0 {
t.Fatalf("precondition: expected a persisted ci-failed tracker")
}
// Approved+green ends the incident → recovered() clears every tracker.
if err := m.ApplySCMObservation(ctx(), sid, ports.SCMFacts{
Fetched: true, PRState: domain.PROpen, ReviewDecision: ports.ReviewApproved, CISummary: ports.CIPassing, PRNumber: 7,
}); err != nil {
t.Fatalf("recover: %v", err)
}
if rows, _ := store.ListReactionTrackers(context.Background()); len(rows) != 0 {
t.Errorf("incident-over must clear durable trackers, got %d rows", len(rows))
}
}

View File

@ -0,0 +1,94 @@
package lifecycle
// reaction_store.go is the optional durability seam for the escalation engine.
// By default the Manager keeps escalation budgets in memory only (a restart
// resets them, which costs at most a few extra agent retries — never a missed
// human page). When a ReactionStore is wired via WithReactionStore the in-memory
// map becomes a write-through cache over durable rows, so a restart does NOT
// re-fire an already-escalated human notification.
//
// The interface uses lifecycle-local types so the package stays free of any
// storage dependency; the composition root adapts the concrete store to it
// (mirroring the cdc.OutboxStore adapter).
import (
"context"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
)
// PersistedTracker is the durable form of one (session,reaction) escalation
// budget — the storage-facing mirror of the in-memory reactionTracker.
type PersistedTracker struct {
SessionID domain.SessionID
Key string
Attempts int
Escalated bool
FirstAttemptAt time.Time
ProjectID domain.ProjectID
}
// ReactionStore persists escalation budgets so they survive a daemon restart.
type ReactionStore interface {
LoadReactionTrackers(ctx context.Context) ([]PersistedTracker, error)
SaveReactionTracker(ctx context.Context, t PersistedTracker) error
DeleteReactionTracker(ctx context.Context, id domain.SessionID, key string) error
DeleteSessionReactionTrackers(ctx context.Context, id domain.SessionID) error
}
// WithReactionStore makes escalation budgets durable: it hydrates the in-memory
// trackers from rs and turns on write-through for subsequent mutations. Like
// WithSessionLister it must be called BEFORE any reaper or Apply* dispatch
// starts, since it populates the tracker map without holding trackerMu against
// concurrent reactors. A hydration error is returned so the caller can decide
// whether to proceed with an empty (in-memory) budget set.
func (m *Manager) WithReactionStore(ctx context.Context, rs ReactionStore) error {
m.reactionStore = rs
rows, err := rs.LoadReactionTrackers(ctx)
if err != nil {
return err
}
for _, r := range rows {
m.trackers[trackerKey{id: r.SessionID, key: reactionKey(r.Key)}] = &reactionTracker{
attempts: r.Attempts,
escalated: r.Escalated,
firstAttemptAt: r.FirstAttemptAt,
projectID: r.ProjectID,
}
}
return nil
}
// persistTracker write-throughs one tracker's current state. Best-effort: a
// failed write degrades durability to the in-memory default (a restart may
// re-fire one page), so it must not break the synchronous dispatch path. The
// snapshot is taken by the caller under trackerMu and passed by value here so no
// DB I/O happens while the lock is held.
func (m *Manager) persistTracker(ctx context.Context, id domain.SessionID, key reactionKey, snap reactionTracker) {
if m.reactionStore == nil {
return
}
_ = m.reactionStore.SaveReactionTracker(ctx, PersistedTracker{
SessionID: id,
Key: string(key),
Attempts: snap.attempts,
Escalated: snap.escalated,
FirstAttemptAt: snap.firstAttemptAt,
ProjectID: snap.projectID,
})
}
func (m *Manager) deletePersistedTracker(ctx context.Context, id domain.SessionID, key reactionKey) {
if m.reactionStore == nil {
return
}
_ = m.reactionStore.DeleteReactionTracker(ctx, id, string(key))
}
func (m *Manager) deletePersistedSessionTrackers(ctx context.Context, id domain.SessionID) {
if m.reactionStore == nil {
return
}
_ = m.reactionStore.DeleteSessionReactionTrackers(ctx, id)
}

View File

@ -233,14 +233,14 @@ func (m *Manager) react(ctx context.Context, id domain.SessionID, tr *transition
// transition is typically review_pending->approved (beforeKey empty), so
// clearing only beforeKey would leak the ci-failed tracker and leave its
// escalated=true to silence a future regression. Clear them all.
m.clearSessionTrackers(id)
m.clearSessionTrackers(ctx, id)
case hadBefore && (!hasAfter || changed):
// Within an unresolved open PR: a normal tracker resets when its state is
// left. A persistent one (ci-failed) is NOT cleared here — it must survive
// the ambiguous review_pending limbo (the fail->pending->fail flap, §4.2);
// it only resets via the recovery/incident-over branch above.
if !defaultReactions[beforeKey].persistent {
m.clearTracker(id, beforeKey)
m.clearTracker(ctx, id, beforeKey)
}
}
@ -324,13 +324,21 @@ func (m *Manager) sendToAgent(ctx context.Context, id domain.SessionID, projectI
tk.firstAttemptAt = now
}
tk.attempts++
if shouldEscalate(tk, cfg, now) {
escalateNow := shouldEscalate(tk, cfg, now)
if escalateNow {
tk.escalated = true
m.trackerMu.Unlock()
return m.escalate(ctx, id, tk.projectID, key)
}
snap := *tk
m.trackerMu.Unlock()
// Write through the new budget (incl. escalated) before dispatching, so a
// crash between persist and notify re-fires at most the same page on restart.
m.persistTracker(ctx, id, key, snap)
if escalateNow {
return m.escalate(ctx, id, snap.projectID, key)
}
if err := m.messenger.Send(ctx, id, composeMessage(cfg, rc)); err != nil {
// A delivery failure must not consume escalation budget: roll this
// attempt back so the next relevant transition retries from the same
@ -341,7 +349,9 @@ func (m *Manager) sendToAgent(ctx context.Context, id domain.SessionID, projectI
if freshFirst {
tk.firstAttemptAt = time.Time{}
}
rolled := *tk
m.trackerMu.Unlock()
m.persistTracker(ctx, id, key, rolled)
return err
}
return nil
@ -393,16 +403,17 @@ func (m *Manager) trackerFor(id domain.SessionID, key reactionKey) *reactionTrac
return tk
}
func (m *Manager) clearTracker(id domain.SessionID, key reactionKey) {
func (m *Manager) clearTracker(ctx context.Context, id domain.SessionID, key reactionKey) {
m.trackerMu.Lock()
delete(m.trackers, trackerKey{id: id, key: key})
m.trackerMu.Unlock()
m.deletePersistedTracker(ctx, id, key)
}
// clearSessionTrackers drops every tracker for a session — used when its
// incident is over, so no budget (and no stale escalated=true) survives into a
// later unrelated incident.
func (m *Manager) clearSessionTrackers(id domain.SessionID) {
func (m *Manager) clearSessionTrackers(ctx context.Context, id domain.SessionID) {
m.trackerMu.Lock()
for k := range m.trackers {
if k.id == id {
@ -410,6 +421,7 @@ func (m *Manager) clearSessionTrackers(id domain.SessionID) {
}
}
m.trackerMu.Unlock()
m.deletePersistedSessionTrackers(ctx, id)
}
// TickEscalations fires the duration-based escalations the synchronous LCM
@ -421,6 +433,7 @@ func (m *Manager) TickEscalations(ctx context.Context, now time.Time) error {
id domain.SessionID
projectID domain.ProjectID
key reactionKey
snap reactionTracker
}
var fire []due
@ -432,12 +445,13 @@ func (m *Manager) TickEscalations(ctx context.Context, now time.Time) error {
cfg := defaultReactions[k.key]
if cfg.escalateAfter > 0 && !tk.firstAttemptAt.IsZero() && now.Sub(tk.firstAttemptAt) >= cfg.escalateAfter {
tk.escalated = true
fire = append(fire, due{id: k.id, projectID: tk.projectID, key: k.key})
fire = append(fire, due{id: k.id, projectID: tk.projectID, key: k.key, snap: *tk})
}
}
m.trackerMu.Unlock()
for _, d := range fire {
m.persistTracker(ctx, d.id, d.key, d.snap)
if err := m.escalate(ctx, d.id, d.projectID, d.key); err != nil {
return err
}

View File

@ -0,0 +1,104 @@
package sqlite
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
)
// OutboxEvent is a single undelivered change, joined from outbox + change_log.
// It is the unit the CDC publisher drains to JSONL.
type OutboxEvent struct {
OutboxID int64
Seq int64
SessionID string
EventType string
Revision int64
Payload string
CreatedAt time.Time
}
// ListUnsent returns up to limit undelivered events in seq order.
func (s *Store) ListUnsent(ctx context.Context, limit int) ([]OutboxEvent, error) {
rows, err := s.q.ListUnsentOutbox(ctx, int64(limit))
if err != nil {
return nil, fmt.Errorf("list unsent outbox: %w", err)
}
out := make([]OutboxEvent, 0, len(rows))
for _, r := range rows {
out = append(out, OutboxEvent{
OutboxID: r.ID,
Seq: r.ChangeLogSeq,
SessionID: r.SessionID,
EventType: r.EventType,
Revision: r.Revision,
Payload: r.Payload,
CreatedAt: r.CreatedAt,
})
}
return out, nil
}
// MarkSent flags an outbox row delivered.
func (s *Store) MarkSent(ctx context.Context, outboxID int64, at time.Time) error {
return s.q.MarkOutboxSent(ctx, gen.MarkOutboxSentParams{
SentAt: sql.NullTime{Time: at, Valid: true},
ID: outboxID,
})
}
// MarkFailed bumps the attempt count and records the last error for an outbox row.
func (s *Store) MarkFailed(ctx context.Context, outboxID int64, errMsg string) error {
return s.q.MarkOutboxFailed(ctx, gen.MarkOutboxFailedParams{LastError: errMsg, ID: outboxID})
}
// GetOffset returns a consumer's last acknowledged seq (0 if it has none).
func (s *Store) GetOffset(ctx context.Context, consumer string) (int64, error) {
seq, err := s.q.GetConsumerOffset(ctx, consumer)
if errors.Is(err, sql.ErrNoRows) {
return 0, nil
}
if err != nil {
return 0, fmt.Errorf("get consumer offset %s: %w", consumer, err)
}
return seq, nil
}
// SetOffset durably records a consumer's acknowledged seq.
func (s *Store) SetOffset(ctx context.Context, consumer string, seq int64, at time.Time) error {
return s.q.UpsertConsumerOffset(ctx, gen.UpsertConsumerOffsetParams{
Consumer: consumer,
LastSeq: seq,
UpdatedAt: at,
})
}
// MaxChangeLogSeq returns the highest change_log seq (0 if empty). Used by the
// consumer to resume after a snapshot resync.
func (s *Store) MaxChangeLogSeq(ctx context.Context) (int64, error) {
v, err := s.q.MaxChangeLogSeq(ctx)
if err != nil {
return 0, fmt.Errorf("max change_log seq: %w", err)
}
return v, nil
}
// MinConsumerOffset returns the lowest acknowledged seq across all consumers
// (0 if none). The janitor uses it as the safe outbox-deletion watermark.
func (s *Store) MinConsumerOffset(ctx context.Context) (int64, error) {
v, err := s.q.MinConsumerOffset(ctx)
if err != nil {
return 0, fmt.Errorf("min consumer offset: %w", err)
}
return v, nil
}
// DeleteSentOutboxBelow removes delivered outbox rows whose seq is below the
// watermark, returning the number removed.
func (s *Store) DeleteSentOutboxBelow(ctx context.Context, seq int64) (int64, error) {
return s.q.DeleteSentOutboxBelow(ctx, seq)
}

View File

@ -0,0 +1,63 @@
// Package sqlite is the durable persistence adapter behind ports.LifecycleStore.
// It owns the SQLite schema (goose migrations), the revision-CAS upsert, and the
// transactional outbox (one txn writes the session row, a change_log entry, and
// the outbox row that the CDC publisher later drains to JSONL).
package sqlite
import (
"database/sql"
"embed"
"fmt"
"os"
"path/filepath"
"github.com/pressly/goose/v3"
_ "modernc.org/sqlite"
)
//go:embed migrations/*.sql
var migrationsFS embed.FS
// pragmas are applied on every connection open. WAL + NORMAL gives concurrent
// reads alongside the single writer; busy_timeout absorbs brief writer
// contention; foreign_keys enforces the session_metadata cascade.
const pragmas = "?_pragma=journal_mode(WAL)" +
"&_pragma=busy_timeout(5000)" +
"&_pragma=foreign_keys(ON)" +
"&_pragma=synchronous(NORMAL)"
// Open opens (creating if absent) the SQLite database under dataDir, applies the
// connection pragmas, and runs all goose migrations up. The returned *sql.DB is
// safe for the single-writer / many-reader workload the LCM and readers impose.
func Open(dataDir string) (*sql.DB, error) {
if err := os.MkdirAll(dataDir, 0o755); err != nil {
return nil, fmt.Errorf("create data dir: %w", err)
}
dsn := "file:" + filepath.Join(dataDir, "ao.db") + pragmas
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, fmt.Errorf("open sqlite: %w", err)
}
// Single writer: serialize all access through one connection so WAL's
// single-writer rule is never violated by the pool handing out a second
// writable conn mid-transaction.
db.SetMaxOpenConns(1)
if err := migrate(db); err != nil {
db.Close()
return nil, err
}
return db, nil
}
func migrate(db *sql.DB) error {
goose.SetBaseFS(migrationsFS)
goose.SetLogger(goose.NopLogger())
if err := goose.SetDialect("sqlite3"); err != nil {
return fmt.Errorf("set goose dialect: %w", err)
}
if err := goose.Up(db, "migrations"); err != nil {
return fmt.Errorf("run migrations: %w", err)
}
return nil
}

View File

@ -0,0 +1,199 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: cdc.sql
package gen
import (
"context"
"database/sql"
"time"
)
const deleteSentOutboxBelow = `-- name: DeleteSentOutboxBelow :execrows
DELETE FROM outbox WHERE sent = 1 AND change_log_seq < ?
`
func (q *Queries) DeleteSentOutboxBelow(ctx context.Context, changeLogSeq int64) (int64, error) {
result, err := q.db.ExecContext(ctx, deleteSentOutboxBelow, changeLogSeq)
if err != nil {
return 0, err
}
return result.RowsAffected()
}
const getConsumerOffset = `-- name: GetConsumerOffset :one
SELECT last_seq FROM consumer_offsets WHERE consumer = ?
`
func (q *Queries) GetConsumerOffset(ctx context.Context, consumer string) (int64, error) {
row := q.db.QueryRowContext(ctx, getConsumerOffset, consumer)
var last_seq int64
err := row.Scan(&last_seq)
return last_seq, err
}
const insertChangeLog = `-- name: InsertChangeLog :one
INSERT INTO change_log (session_id, event_type, revision, payload, created_at)
VALUES (?, ?, ?, ?, ?)
RETURNING seq
`
type InsertChangeLogParams struct {
SessionID string
EventType string
Revision int64
Payload string
CreatedAt time.Time
}
// Appends a canonical-write record and returns its monotonic seq so the same
// transaction can thread it into the outbox row.
func (q *Queries) InsertChangeLog(ctx context.Context, arg InsertChangeLogParams) (int64, error) {
row := q.db.QueryRowContext(ctx, insertChangeLog,
arg.SessionID,
arg.EventType,
arg.Revision,
arg.Payload,
arg.CreatedAt,
)
var seq int64
err := row.Scan(&seq)
return seq, err
}
const insertOutbox = `-- name: InsertOutbox :exec
INSERT INTO outbox (change_log_seq, created_at)
VALUES (?, ?)
`
type InsertOutboxParams struct {
ChangeLogSeq int64
CreatedAt time.Time
}
func (q *Queries) InsertOutbox(ctx context.Context, arg InsertOutboxParams) error {
_, err := q.db.ExecContext(ctx, insertOutbox, arg.ChangeLogSeq, arg.CreatedAt)
return err
}
const listUnsentOutbox = `-- name: ListUnsentOutbox :many
SELECT o.id, o.change_log_seq, o.attempts,
c.session_id, c.event_type, c.revision, c.payload, c.created_at
FROM outbox o
JOIN change_log c ON c.seq = o.change_log_seq
WHERE o.sent = 0
ORDER BY o.change_log_seq
LIMIT ?
`
type ListUnsentOutboxRow struct {
ID int64
ChangeLogSeq int64
Attempts int64
SessionID string
EventType string
Revision int64
Payload string
CreatedAt time.Time
}
func (q *Queries) ListUnsentOutbox(ctx context.Context, limit int64) ([]ListUnsentOutboxRow, error) {
rows, err := q.db.QueryContext(ctx, listUnsentOutbox, limit)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListUnsentOutboxRow{}
for rows.Next() {
var i ListUnsentOutboxRow
if err := rows.Scan(
&i.ID,
&i.ChangeLogSeq,
&i.Attempts,
&i.SessionID,
&i.EventType,
&i.Revision,
&i.Payload,
&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 markOutboxFailed = `-- name: MarkOutboxFailed :exec
UPDATE outbox SET attempts = attempts + 1, last_error = ? WHERE id = ?
`
type MarkOutboxFailedParams struct {
LastError string
ID int64
}
func (q *Queries) MarkOutboxFailed(ctx context.Context, arg MarkOutboxFailedParams) error {
_, err := q.db.ExecContext(ctx, markOutboxFailed, arg.LastError, arg.ID)
return err
}
const markOutboxSent = `-- name: MarkOutboxSent :exec
UPDATE outbox SET sent = 1, sent_at = ? WHERE id = ?
`
type MarkOutboxSentParams struct {
SentAt sql.NullTime
ID int64
}
func (q *Queries) MarkOutboxSent(ctx context.Context, arg MarkOutboxSentParams) error {
_, err := q.db.ExecContext(ctx, markOutboxSent, arg.SentAt, arg.ID)
return err
}
const maxChangeLogSeq = `-- name: MaxChangeLogSeq :one
SELECT CAST(COALESCE(MAX(seq), 0) AS INTEGER) FROM change_log
`
func (q *Queries) MaxChangeLogSeq(ctx context.Context) (int64, error) {
row := q.db.QueryRowContext(ctx, maxChangeLogSeq)
var column_1 int64
err := row.Scan(&column_1)
return column_1, err
}
const minConsumerOffset = `-- name: MinConsumerOffset :one
SELECT CAST(COALESCE(MIN(last_seq), 0) AS INTEGER) FROM consumer_offsets
`
func (q *Queries) MinConsumerOffset(ctx context.Context) (int64, error) {
row := q.db.QueryRowContext(ctx, minConsumerOffset)
var column_1 int64
err := row.Scan(&column_1)
return column_1, err
}
const upsertConsumerOffset = `-- name: UpsertConsumerOffset :exec
INSERT INTO consumer_offsets (consumer, last_seq, updated_at)
VALUES (?, ?, ?)
ON CONFLICT (consumer) DO UPDATE SET last_seq = excluded.last_seq, updated_at = excluded.updated_at
`
type UpsertConsumerOffsetParams struct {
Consumer string
LastSeq int64
UpdatedAt time.Time
}
func (q *Queries) UpsertConsumerOffset(ctx context.Context, arg UpsertConsumerOffsetParams) error {
_, err := q.db.ExecContext(ctx, upsertConsumerOffset, arg.Consumer, arg.LastSeq, arg.UpdatedAt)
return err
}

View File

@ -0,0 +1,31 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
package gen
import (
"context"
"database/sql"
)
type DBTX interface {
ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
PrepareContext(context.Context, string) (*sql.Stmt, error)
QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
QueryRowContext(context.Context, string, ...interface{}) *sql.Row
}
func New(db DBTX) *Queries {
return &Queries{db: db}
}
type Queries struct {
db DBTX
}
func (q *Queries) WithTx(tx *sql.Tx) *Queries {
return &Queries{
db: tx,
}
}

View File

@ -0,0 +1,59 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: metadata.sql
package gen
import (
"context"
)
const getMetadata = `-- name: GetMetadata :many
SELECT key, value FROM session_metadata WHERE session_id = ?
`
type GetMetadataRow struct {
Key string
Value 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
}
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
`
type UpsertMetadataParams struct {
SessionID string
Key string
Value string
}
func (q *Queries) UpsertMetadata(ctx context.Context, arg UpsertMetadataParams) error {
_, err := q.db.ExecContext(ctx, upsertMetadata, arg.SessionID, arg.Key, arg.Value)
return err
}

View File

@ -0,0 +1,74 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
package gen
import (
"database/sql"
"time"
)
type ChangeLog struct {
Seq int64
SessionID string
EventType string
Revision int64
Payload string
CreatedAt time.Time
}
type ConsumerOffset struct {
Consumer string
LastSeq int64
UpdatedAt time.Time
}
type Outbox struct {
ID int64
ChangeLogSeq int64
Sent int64
SentAt sql.NullTime
Attempts int64
LastError string
CreatedAt time.Time
}
type ReactionTracker struct {
SessionID string
ReactionKey string
Attempts int64
Escalated int64
FirstAttemptAt sql.NullTime
ProjectID string
}
type Session struct {
ID string
ProjectID string
IssueID string
Kind string
CreatedAt time.Time
UpdatedAt time.Time
Revision int64
SessionState string
SessionReason string
PrState string
PrReason string
PrNumber int64
PrUrl string
RuntimeState string
RuntimeReason string
ActivityState string
ActivityLastAt time.Time
ActivitySource string
DetectingAttempts sql.NullInt64
DetectingStartedAt sql.NullTime
DetectingEvidenceHash sql.NullString
}
type SessionMetadatum struct {
SessionID string
Key string
Value string
}

View File

@ -0,0 +1,42 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
package gen
import (
"context"
)
type Querier interface {
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)
GetSession(ctx context.Context, id string) (Session, 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
// 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)
ListReactionTrackers(ctx context.Context) ([]ReactionTracker, error)
ListSessionsByProject(ctx context.Context, projectID string) ([]Session, error)
ListUnsentOutbox(ctx context.Context, limit int64) ([]ListUnsentOutboxRow, error)
MarkOutboxFailed(ctx context.Context, arg MarkOutboxFailedParams) error
MarkOutboxSent(ctx context.Context, arg MarkOutboxSentParams) error
MaxChangeLogSeq(ctx context.Context) (int64, error)
MinConsumerOffset(ctx context.Context) (int64, error)
// CAS update: succeeds only when the stored revision equals the caller's loaded
// 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
UpsertReactionTracker(ctx context.Context, arg UpsertReactionTrackerParams) error
}
var _ Querier = (*Queries)(nil)

View File

@ -0,0 +1,100 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: reactions.sql
package gen
import (
"context"
"database/sql"
)
const deleteReactionTracker = `-- name: DeleteReactionTracker :exec
DELETE FROM reaction_trackers WHERE session_id = ? AND reaction_key = ?
`
type DeleteReactionTrackerParams struct {
SessionID string
ReactionKey string
}
func (q *Queries) DeleteReactionTracker(ctx context.Context, arg DeleteReactionTrackerParams) error {
_, err := q.db.ExecContext(ctx, deleteReactionTracker, arg.SessionID, arg.ReactionKey)
return err
}
const deleteSessionReactionTrackers = `-- name: DeleteSessionReactionTrackers :exec
DELETE FROM reaction_trackers WHERE session_id = ?
`
func (q *Queries) DeleteSessionReactionTrackers(ctx context.Context, sessionID string) error {
_, err := q.db.ExecContext(ctx, deleteSessionReactionTrackers, sessionID)
return err
}
const listReactionTrackers = `-- name: ListReactionTrackers :many
SELECT session_id, reaction_key, attempts, escalated, first_attempt_at, project_id
FROM reaction_trackers
`
func (q *Queries) ListReactionTrackers(ctx context.Context) ([]ReactionTracker, error) {
rows, err := q.db.QueryContext(ctx, listReactionTrackers)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ReactionTracker{}
for rows.Next() {
var i ReactionTracker
if err := rows.Scan(
&i.SessionID,
&i.ReactionKey,
&i.Attempts,
&i.Escalated,
&i.FirstAttemptAt,
&i.ProjectID,
); 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 upsertReactionTracker = `-- name: UpsertReactionTracker :exec
INSERT INTO reaction_trackers (session_id, reaction_key, attempts, escalated, first_attempt_at, project_id)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT (session_id, reaction_key) DO UPDATE SET
attempts = excluded.attempts,
escalated = excluded.escalated,
first_attempt_at = excluded.first_attempt_at,
project_id = excluded.project_id
`
type UpsertReactionTrackerParams struct {
SessionID string
ReactionKey string
Attempts int64
Escalated int64
FirstAttemptAt sql.NullTime
ProjectID string
}
func (q *Queries) UpsertReactionTracker(ctx context.Context, arg UpsertReactionTrackerParams) error {
_, err := q.db.ExecContext(ctx, upsertReactionTracker,
arg.SessionID,
arg.ReactionKey,
arg.Attempts,
arg.Escalated,
arg.FirstAttemptAt,
arg.ProjectID,
)
return err
}

View File

@ -0,0 +1,307 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: sessions.sql
package gen
import (
"context"
"database/sql"
"time"
)
const getSession = `-- name: GetSession :one
SELECT id, project_id, issue_id, kind, created_at, updated_at, revision, session_state, session_reason, pr_state, pr_reason, pr_number, pr_url, runtime_state, runtime_reason, activity_state, activity_last_at, activity_source, detecting_attempts, detecting_started_at, detecting_evidence_hash FROM sessions WHERE id = ?
`
func (q *Queries) GetSession(ctx context.Context, id string) (Session, error) {
row := q.db.QueryRowContext(ctx, getSession, id)
var i Session
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.IssueID,
&i.Kind,
&i.CreatedAt,
&i.UpdatedAt,
&i.Revision,
&i.SessionState,
&i.SessionReason,
&i.PrState,
&i.PrReason,
&i.PrNumber,
&i.PrUrl,
&i.RuntimeState,
&i.RuntimeReason,
&i.ActivityState,
&i.ActivityLastAt,
&i.ActivitySource,
&i.DetectingAttempts,
&i.DetectingStartedAt,
&i.DetectingEvidenceHash,
)
return i, err
}
const getSessionRevision = `-- name: GetSessionRevision :one
SELECT revision FROM sessions WHERE id = ?
`
func (q *Queries) GetSessionRevision(ctx context.Context, id string) (int64, error) {
row := q.db.QueryRowContext(ctx, getSessionRevision, id)
var revision int64
err := row.Scan(&revision)
return revision, err
}
const insertSession = `-- name: InsertSession :execrows
INSERT INTO sessions (
id, project_id, issue_id, kind, created_at, updated_at,
revision,
session_state, session_reason,
pr_state, pr_reason, pr_number, pr_url,
runtime_state, runtime_reason,
activity_state, activity_last_at, activity_source,
detecting_attempts, detecting_started_at, detecting_evidence_hash
) VALUES (
?, ?, ?, ?, ?, ?,
1,
?, ?,
?, ?, ?, ?,
?, ?,
?, ?, ?,
?, ?, ?
)
ON CONFLICT (id) DO NOTHING
`
type InsertSessionParams struct {
ID string
ProjectID string
IssueID string
Kind string
CreatedAt time.Time
UpdatedAt time.Time
SessionState string
SessionReason string
PrState string
PrReason string
PrNumber int64
PrUrl string
RuntimeState string
RuntimeReason string
ActivityState string
ActivityLastAt time.Time
ActivitySource string
DetectingAttempts sql.NullInt64
DetectingStartedAt sql.NullTime
DetectingEvidenceHash sql.NullString
}
// CAS insert: only succeeds for a brand-new id. Incoming revision must be 0;
// the row is persisted at revision 1.
func (q *Queries) InsertSession(ctx context.Context, arg InsertSessionParams) (int64, error) {
result, err := q.db.ExecContext(ctx, insertSession,
arg.ID,
arg.ProjectID,
arg.IssueID,
arg.Kind,
arg.CreatedAt,
arg.UpdatedAt,
arg.SessionState,
arg.SessionReason,
arg.PrState,
arg.PrReason,
arg.PrNumber,
arg.PrUrl,
arg.RuntimeState,
arg.RuntimeReason,
arg.ActivityState,
arg.ActivityLastAt,
arg.ActivitySource,
arg.DetectingAttempts,
arg.DetectingStartedAt,
arg.DetectingEvidenceHash,
)
if err != nil {
return 0, err
}
return result.RowsAffected()
}
const listAllSessions = `-- name: ListAllSessions :many
SELECT id, project_id, issue_id, kind, created_at, updated_at, revision, session_state, session_reason, pr_state, pr_reason, pr_number, pr_url, runtime_state, runtime_reason, activity_state, activity_last_at, activity_source, detecting_attempts, detecting_started_at, detecting_evidence_hash FROM sessions
`
func (q *Queries) ListAllSessions(ctx context.Context) ([]Session, error) {
rows, err := q.db.QueryContext(ctx, listAllSessions)
if err != nil {
return nil, err
}
defer rows.Close()
items := []Session{}
for rows.Next() {
var i Session
if err := rows.Scan(
&i.ID,
&i.ProjectID,
&i.IssueID,
&i.Kind,
&i.CreatedAt,
&i.UpdatedAt,
&i.Revision,
&i.SessionState,
&i.SessionReason,
&i.PrState,
&i.PrReason,
&i.PrNumber,
&i.PrUrl,
&i.RuntimeState,
&i.RuntimeReason,
&i.ActivityState,
&i.ActivityLastAt,
&i.ActivitySource,
&i.DetectingAttempts,
&i.DetectingStartedAt,
&i.DetectingEvidenceHash,
); 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 listSessionsByProject = `-- name: ListSessionsByProject :many
SELECT id, project_id, issue_id, kind, created_at, updated_at, revision, session_state, session_reason, pr_state, pr_reason, pr_number, pr_url, runtime_state, runtime_reason, activity_state, activity_last_at, activity_source, detecting_attempts, detecting_started_at, detecting_evidence_hash FROM sessions WHERE project_id = ?
`
func (q *Queries) ListSessionsByProject(ctx context.Context, projectID string) ([]Session, error) {
rows, err := q.db.QueryContext(ctx, listSessionsByProject, projectID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []Session{}
for rows.Next() {
var i Session
if err := rows.Scan(
&i.ID,
&i.ProjectID,
&i.IssueID,
&i.Kind,
&i.CreatedAt,
&i.UpdatedAt,
&i.Revision,
&i.SessionState,
&i.SessionReason,
&i.PrState,
&i.PrReason,
&i.PrNumber,
&i.PrUrl,
&i.RuntimeState,
&i.RuntimeReason,
&i.ActivityState,
&i.ActivityLastAt,
&i.ActivitySource,
&i.DetectingAttempts,
&i.DetectingStartedAt,
&i.DetectingEvidenceHash,
); 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 updateSessionCAS = `-- name: UpdateSessionCAS :execrows
UPDATE sessions SET
project_id = ?,
issue_id = ?,
kind = ?,
updated_at = ?,
revision = revision + 1,
session_state = ?,
session_reason = ?,
pr_state = ?,
pr_reason = ?,
pr_number = ?,
pr_url = ?,
runtime_state = ?,
runtime_reason = ?,
activity_state = ?,
activity_last_at = ?,
activity_source = ?,
detecting_attempts = ?,
detecting_started_at = ?,
detecting_evidence_hash = ?
WHERE id = ? AND revision = ?
`
type UpdateSessionCASParams struct {
ProjectID string
IssueID string
Kind string
UpdatedAt time.Time
SessionState string
SessionReason string
PrState string
PrReason string
PrNumber int64
PrUrl string
RuntimeState string
RuntimeReason string
ActivityState string
ActivityLastAt time.Time
ActivitySource string
DetectingAttempts sql.NullInt64
DetectingStartedAt sql.NullTime
DetectingEvidenceHash sql.NullString
ID string
Revision int64
}
// CAS update: succeeds only when the stored revision equals the caller's loaded
// revision (@expected_revision). 0 rows affected => revision mismatch.
func (q *Queries) UpdateSessionCAS(ctx context.Context, arg UpdateSessionCASParams) (int64, error) {
result, err := q.db.ExecContext(ctx, updateSessionCAS,
arg.ProjectID,
arg.IssueID,
arg.Kind,
arg.UpdatedAt,
arg.SessionState,
arg.SessionReason,
arg.PrState,
arg.PrReason,
arg.PrNumber,
arg.PrUrl,
arg.RuntimeState,
arg.RuntimeReason,
arg.ActivityState,
arg.ActivityLastAt,
arg.ActivitySource,
arg.DetectingAttempts,
arg.DetectingStartedAt,
arg.DetectingEvidenceHash,
arg.ID,
arg.Revision,
)
if err != nil {
return 0, err
}
return result.RowsAffected()
}

View File

@ -0,0 +1,129 @@
package sqlite
import (
"database/sql"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
)
// recordToInsert maps a domain record to the generated insert params. The
// revision column is fixed to 1 by the query itself (insert path), so it is not
// carried here.
func recordToInsert(rec domain.SessionRecord) gen.InsertSessionParams {
lc := rec.Lifecycle
da, ds, dh := detectingToNull(lc.Detecting)
return gen.InsertSessionParams{
ID: string(rec.ID),
ProjectID: string(rec.ProjectID),
IssueID: string(rec.IssueID),
Kind: string(rec.Kind),
CreatedAt: rec.CreatedAt,
UpdatedAt: rec.UpdatedAt,
SessionState: string(lc.Session.State),
SessionReason: string(lc.Session.Reason),
PrState: string(lc.PR.State),
PrReason: string(lc.PR.Reason),
PrNumber: int64(lc.PR.Number),
PrUrl: lc.PR.URL,
RuntimeState: string(lc.Runtime.State),
RuntimeReason: string(lc.Runtime.Reason),
ActivityState: string(lc.Activity.State),
ActivityLastAt: lc.Activity.LastActivityAt,
ActivitySource: string(lc.Activity.Source),
DetectingAttempts: da,
DetectingStartedAt: ds,
DetectingEvidenceHash: dh,
}
}
// recordToUpdate maps a domain record to the CAS update params. expectedRevision
// is the caller's loaded revision, used in the WHERE clause for the CAS check.
func recordToUpdate(rec domain.SessionRecord, expectedRevision int64) gen.UpdateSessionCASParams {
lc := rec.Lifecycle
da, ds, dh := detectingToNull(lc.Detecting)
return gen.UpdateSessionCASParams{
ProjectID: string(rec.ProjectID),
IssueID: string(rec.IssueID),
Kind: string(rec.Kind),
UpdatedAt: rec.UpdatedAt,
SessionState: string(lc.Session.State),
SessionReason: string(lc.Session.Reason),
PrState: string(lc.PR.State),
PrReason: string(lc.PR.Reason),
PrNumber: int64(lc.PR.Number),
PrUrl: lc.PR.URL,
RuntimeState: string(lc.Runtime.State),
RuntimeReason: string(lc.Runtime.Reason),
ActivityState: string(lc.Activity.State),
ActivityLastAt: lc.Activity.LastActivityAt,
ActivitySource: string(lc.Activity.Source),
DetectingAttempts: da,
DetectingStartedAt: ds,
DetectingEvidenceHash: dh,
ID: string(rec.ID),
Revision: expectedRevision,
}
}
// rowToRecord maps a stored session row back to a domain record. Metadata is
// deliberately left nil: it is a side-channel (session_metadata) read only by
// GetMetadata, never reconstructed here — mirroring the in-memory fakeStore.
func rowToRecord(row gen.Session) domain.SessionRecord {
return domain.SessionRecord{
ID: domain.SessionID(row.ID),
ProjectID: domain.ProjectID(row.ProjectID),
IssueID: domain.IssueID(row.IssueID),
Kind: domain.SessionKind(row.Kind),
Lifecycle: rowToLifecycle(row),
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
}
}
func rowToLifecycle(row gen.Session) domain.CanonicalSessionLifecycle {
return domain.CanonicalSessionLifecycle{
Version: domain.LifecycleVersion,
Revision: int(row.Revision),
Session: domain.SessionSubstate{
State: domain.SessionState(row.SessionState),
Reason: domain.SessionReason(row.SessionReason),
},
PR: domain.PRSubstate{
State: domain.PRState(row.PrState),
Reason: domain.PRReason(row.PrReason),
Number: int(row.PrNumber),
URL: row.PrUrl,
},
Runtime: domain.RuntimeSubstate{
State: domain.RuntimeState(row.RuntimeState),
Reason: domain.RuntimeReason(row.RuntimeReason),
},
Activity: domain.ActivitySubstate{
State: domain.ActivityState(row.ActivityState),
LastActivityAt: row.ActivityLastAt,
Source: domain.ActivitySource(row.ActivitySource),
},
Detecting: nullToDetecting(row),
}
}
func detectingToNull(d *domain.DetectingState) (sql.NullInt64, sql.NullTime, sql.NullString) {
if d == nil {
return sql.NullInt64{}, sql.NullTime{}, sql.NullString{}
}
return sql.NullInt64{Int64: int64(d.Attempts), Valid: true},
sql.NullTime{Time: d.StartedAt, Valid: true},
sql.NullString{String: d.EvidenceHash, Valid: true}
}
func nullToDetecting(row gen.Session) *domain.DetectingState {
if !row.DetectingAttempts.Valid {
return nil
}
return &domain.DetectingState{
Attempts: int(row.DetectingAttempts.Int64),
StartedAt: row.DetectingStartedAt.Time,
EvidenceHash: row.DetectingEvidenceHash.String,
}
}

View File

@ -0,0 +1,109 @@
-- +goose Up
-- +goose StatementBegin
-- sessions holds identity + the canonical lifecycle as typed columns. The
-- display status is NEVER stored (it is derived on read). Metadata is NOT here —
-- it lives in session_metadata, written by a side-channel that bypasses CDC.
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
project_id TEXT NOT NULL,
issue_id TEXT NOT NULL DEFAULT '',
kind TEXT NOT NULL,
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL,
-- canonical lifecycle: revision is the optimistic-concurrency (CAS) counter,
-- bumped only by the storage layer's Upsert.
revision INTEGER NOT NULL,
session_state TEXT NOT NULL,
session_reason TEXT NOT NULL,
pr_state TEXT NOT NULL,
pr_reason TEXT NOT NULL,
pr_number INTEGER NOT NULL DEFAULT 0,
pr_url TEXT NOT NULL DEFAULT '',
runtime_state TEXT NOT NULL,
runtime_reason TEXT NOT NULL,
activity_state TEXT NOT NULL,
activity_last_at TIMESTAMP NOT NULL,
activity_source TEXT NOT NULL,
-- detecting quarantine memory; NULL when the session is not in detecting.
detecting_attempts INTEGER,
detecting_started_at TIMESTAMP,
detecting_evidence_hash TEXT
);
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.
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)
);
-- change_log is the durable, ordered record of every canonical write. seq is the
-- monotonic CDC ordering/idempotency key.
CREATE TABLE change_log (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
event_type TEXT NOT NULL,
revision INTEGER NOT NULL,
payload TEXT NOT NULL,
created_at TIMESTAMP NOT NULL
);
-- outbox is the transactional-outbox: one unsent row per canonical write, drained
-- by the publisher into JSONL. change_log_seq links it to its change_log row.
CREATE TABLE outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
change_log_seq INTEGER NOT NULL REFERENCES change_log (seq),
sent INTEGER NOT NULL DEFAULT 0,
sent_at TIMESTAMP,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT NOT NULL DEFAULT '',
created_at TIMESTAMP NOT NULL
);
CREATE INDEX idx_outbox_unsent ON outbox (change_log_seq) WHERE sent = 0;
-- consumer_offsets is the durable per-consumer cursor (at-least-once delivery).
CREATE TABLE consumer_offsets (
consumer TEXT PRIMARY KEY,
last_seq INTEGER NOT NULL DEFAULT 0,
updated_at TIMESTAMP NOT NULL
);
-- reaction_trackers is the durable escalation budget (persisted so a restart does
-- not re-fire human pages). Off the canonical CDC path. Mirrors the LCM's
-- in-memory reactionTracker: attempts (numeric budget), escalated (silences
-- further auto-dispatch), first_attempt_at (duration-escalation anchor),
-- project_id (captured at first attempt for the escalation event).
CREATE TABLE reaction_trackers (
session_id TEXT NOT NULL,
reaction_key TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
escalated INTEGER NOT NULL DEFAULT 0,
first_attempt_at TIMESTAMP,
project_id TEXT NOT NULL DEFAULT '',
PRIMARY KEY (session_id, reaction_key)
);
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
DROP TABLE reaction_trackers;
DROP TABLE consumer_offsets;
DROP TABLE outbox;
DROP TABLE change_log;
DROP TABLE session_metadata;
DROP TABLE sessions;
-- +goose StatementEnd

View File

@ -0,0 +1,42 @@
-- name: InsertChangeLog :one
-- Appends a canonical-write record and returns its monotonic seq so the same
-- transaction can thread it into the outbox row.
INSERT INTO change_log (session_id, event_type, revision, payload, created_at)
VALUES (?, ?, ?, ?, ?)
RETURNING seq;
-- name: InsertOutbox :exec
INSERT INTO outbox (change_log_seq, created_at)
VALUES (?, ?);
-- name: ListUnsentOutbox :many
SELECT o.id, o.change_log_seq, o.attempts,
c.session_id, c.event_type, c.revision, c.payload, c.created_at
FROM outbox o
JOIN change_log c ON c.seq = o.change_log_seq
WHERE o.sent = 0
ORDER BY o.change_log_seq
LIMIT ?;
-- name: MarkOutboxSent :exec
UPDATE outbox SET sent = 1, sent_at = ? WHERE id = ?;
-- name: MarkOutboxFailed :exec
UPDATE outbox SET attempts = attempts + 1, last_error = ? WHERE id = ?;
-- name: GetConsumerOffset :one
SELECT last_seq FROM consumer_offsets WHERE consumer = ?;
-- name: UpsertConsumerOffset :exec
INSERT INTO consumer_offsets (consumer, last_seq, updated_at)
VALUES (?, ?, ?)
ON CONFLICT (consumer) DO UPDATE SET last_seq = excluded.last_seq, updated_at = excluded.updated_at;
-- name: MaxChangeLogSeq :one
SELECT CAST(COALESCE(MAX(seq), 0) AS INTEGER) FROM change_log;
-- name: MinConsumerOffset :one
SELECT CAST(COALESCE(MIN(last_seq), 0) AS INTEGER) FROM consumer_offsets;
-- name: DeleteSentOutboxBelow :execrows
DELETE FROM outbox WHERE sent = 1 AND change_log_seq < ?;

View File

@ -0,0 +1,7 @@
-- name: GetMetadata :many
SELECT key, value 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;

View File

@ -0,0 +1,18 @@
-- name: ListReactionTrackers :many
SELECT session_id, reaction_key, attempts, escalated, first_attempt_at, project_id
FROM reaction_trackers;
-- name: UpsertReactionTracker :exec
INSERT INTO reaction_trackers (session_id, reaction_key, attempts, escalated, first_attempt_at, project_id)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT (session_id, reaction_key) DO UPDATE SET
attempts = excluded.attempts,
escalated = excluded.escalated,
first_attempt_at = excluded.first_attempt_at,
project_id = excluded.project_id;
-- name: DeleteReactionTracker :exec
DELETE FROM reaction_trackers WHERE session_id = ? AND reaction_key = ?;
-- name: DeleteSessionReactionTrackers :exec
DELETE FROM reaction_trackers WHERE session_id = ?;

View File

@ -0,0 +1,58 @@
-- name: InsertSession :execrows
-- CAS insert: only succeeds for a brand-new id. Incoming revision must be 0;
-- the row is persisted at revision 1.
INSERT INTO sessions (
id, project_id, issue_id, kind, created_at, updated_at,
revision,
session_state, session_reason,
pr_state, pr_reason, pr_number, pr_url,
runtime_state, runtime_reason,
activity_state, activity_last_at, activity_source,
detecting_attempts, detecting_started_at, detecting_evidence_hash
) VALUES (
?, ?, ?, ?, ?, ?,
1,
?, ?,
?, ?, ?, ?,
?, ?,
?, ?, ?,
?, ?, ?
)
ON CONFLICT (id) DO NOTHING;
-- name: UpdateSessionCAS :execrows
-- CAS update: succeeds only when the stored revision equals the caller's loaded
-- revision (@expected_revision). 0 rows affected => revision mismatch.
UPDATE sessions SET
project_id = ?,
issue_id = ?,
kind = ?,
updated_at = ?,
revision = revision + 1,
session_state = ?,
session_reason = ?,
pr_state = ?,
pr_reason = ?,
pr_number = ?,
pr_url = ?,
runtime_state = ?,
runtime_reason = ?,
activity_state = ?,
activity_last_at = ?,
activity_source = ?,
detecting_attempts = ?,
detecting_started_at = ?,
detecting_evidence_hash = ?
WHERE id = ? AND revision = ?;
-- name: GetSessionRevision :one
SELECT revision FROM sessions WHERE id = ?;
-- name: GetSession :one
SELECT * FROM sessions WHERE id = ?;
-- name: ListSessionsByProject :many
SELECT * FROM sessions WHERE project_id = ?;
-- name: ListAllSessions :many
SELECT * FROM sessions;

View File

@ -0,0 +1,80 @@
package sqlite
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
)
// ReactionTrackerRow is one persisted escalation budget, the durable mirror of
// the LCM's in-memory reactionTracker. It is the unit the lifecycle Manager
// hydrates on startup and writes through on each mutation.
type ReactionTrackerRow struct {
SessionID string
ReactionKey string
Attempts int
Escalated bool
FirstAttemptAt time.Time
ProjectID string
}
// ListReactionTrackers returns every persisted escalation budget so the Manager
// can rehydrate its in-memory trackers after a restart.
func (s *Store) ListReactionTrackers(ctx context.Context) ([]ReactionTrackerRow, error) {
rows, err := s.q.ListReactionTrackers(ctx)
if err != nil {
return nil, fmt.Errorf("list reaction trackers: %w", err)
}
out := make([]ReactionTrackerRow, 0, len(rows))
for _, r := range rows {
var first time.Time
if r.FirstAttemptAt.Valid {
first = r.FirstAttemptAt.Time
}
out = append(out, ReactionTrackerRow{
SessionID: r.SessionID,
ReactionKey: r.ReactionKey,
Attempts: int(r.Attempts),
Escalated: r.Escalated != 0,
FirstAttemptAt: first,
ProjectID: r.ProjectID,
})
}
return out, nil
}
// SaveReactionTracker durably persists one escalation budget (insert or update).
func (s *Store) SaveReactionTracker(ctx context.Context, r ReactionTrackerRow) error {
escalated := int64(0)
if r.Escalated {
escalated = 1
}
first := sql.NullTime{}
if !r.FirstAttemptAt.IsZero() {
first = sql.NullTime{Time: r.FirstAttemptAt, Valid: true}
}
return s.q.UpsertReactionTracker(ctx, gen.UpsertReactionTrackerParams{
SessionID: r.SessionID,
ReactionKey: r.ReactionKey,
Attempts: int64(r.Attempts),
Escalated: escalated,
FirstAttemptAt: first,
ProjectID: r.ProjectID,
})
}
// DeleteReactionTracker drops one escalation budget.
func (s *Store) DeleteReactionTracker(ctx context.Context, sessionID, reactionKey string) error {
return s.q.DeleteReactionTracker(ctx, gen.DeleteReactionTrackerParams{
SessionID: sessionID,
ReactionKey: reactionKey,
})
}
// DeleteSessionReactionTrackers drops every escalation budget for a session.
func (s *Store) DeleteSessionReactionTrackers(ctx context.Context, sessionID string) error {
return s.q.DeleteSessionReactionTrackers(ctx, sessionID)
}

View File

@ -0,0 +1,92 @@
package sqlite
import (
"context"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
)
// TestSpikeOutboxTxn de-risks the whole adapter: it proves the sqlc-generated
// Querier composes inside one *sql.Tx and that the change_log seq returned
// mid-transaction threads into the outbox row — the transactional-outbox shape
// the publisher later drains. Step 0 of the implementation plan.
func TestSpikeOutboxTxn(t *testing.T) {
db, err := Open(t.TempDir())
if err != nil {
t.Fatalf("open: %v", err)
}
defer db.Close()
ctx := context.Background()
now := time.Now().UTC()
tx, err := db.BeginTx(ctx, nil)
if err != nil {
t.Fatalf("begin: %v", err)
}
defer tx.Rollback()
q := gen.New(db).WithTx(tx)
// 1. CAS insert of a brand-new session (revision 0 -> persisted 1).
rows, err := q.InsertSession(ctx, gen.InsertSessionParams{
ID: "s1",
ProjectID: "p1",
Kind: "worker",
CreatedAt: now,
UpdatedAt: now,
SessionState: "working",
SessionReason: "spawn_requested",
PrState: "none",
PrReason: "not_created",
RuntimeState: "unknown",
RuntimeReason: "spawn_incomplete",
ActivityState: "active",
ActivityLastAt: now,
ActivitySource: "none",
})
if err != nil {
t.Fatalf("insert session: %v", err)
}
if rows != 1 {
t.Fatalf("insert session affected %d rows, want 1", rows)
}
// 2. Append the change_log entry and capture its seq mid-transaction.
seq, err := q.InsertChangeLog(ctx, gen.InsertChangeLogParams{
SessionID: "s1",
EventType: "session_created",
Revision: 1,
Payload: `{"id":"s1"}`,
CreatedAt: now,
})
if err != nil {
t.Fatalf("insert change_log: %v", err)
}
if seq != 1 {
t.Fatalf("change_log seq = %d, want 1", seq)
}
// 3. Thread the seq into the outbox row — the key thing the spike validates.
if err := q.InsertOutbox(ctx, gen.InsertOutboxParams{ChangeLogSeq: seq, CreatedAt: now}); err != nil {
t.Fatalf("insert outbox: %v", err)
}
if err := tx.Commit(); err != nil {
t.Fatalf("commit: %v", err)
}
// Verify the outbox row is visible, unsent, and linked to change_log seq 1.
unsent, err := gen.New(db).ListUnsentOutbox(ctx, 10)
if err != nil {
t.Fatalf("list unsent: %v", err)
}
if len(unsent) != 1 {
t.Fatalf("unsent outbox = %d rows, want 1", len(unsent))
}
if unsent[0].ChangeLogSeq != 1 || unsent[0].SessionID != "s1" || unsent[0].EventType != "session_created" {
t.Fatalf("unexpected outbox row: %+v", unsent[0])
}
}

View File

@ -0,0 +1,118 @@
package sqlite
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
)
// Store is the SQLite-backed ports.LifecycleStore. The LCM is its sole logical
// writer (via Upsert); readers (Session Manager, reaper) use Load/Get/List.
type Store struct {
db *sql.DB
q *gen.Queries
}
var _ ports.LifecycleStore = (*Store)(nil)
// NewStore wraps an opened *sql.DB (see Open) as a LifecycleStore.
func NewStore(db *sql.DB) *Store {
return &Store{db: db, q: gen.New(db)}
}
// Load returns the canonical lifecycle for a session, or ok=false if absent.
func (s *Store) Load(ctx context.Context, id domain.SessionID) (domain.CanonicalSessionLifecycle, bool, error) {
row, err := s.q.GetSession(ctx, string(id))
if errors.Is(err, sql.ErrNoRows) {
return domain.CanonicalSessionLifecycle{}, false, nil
}
if err != nil {
return domain.CanonicalSessionLifecycle{}, false, fmt.Errorf("load session %s: %w", id, err)
}
return rowToLifecycle(row), true, nil
}
// Get returns the full record (no derived status) for a session.
func (s *Store) Get(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) {
row, err := s.q.GetSession(ctx, string(id))
if errors.Is(err, sql.ErrNoRows) {
return domain.SessionRecord{}, false, nil
}
if err != nil {
return domain.SessionRecord{}, false, fmt.Errorf("get session %s: %w", id, err)
}
return rowToRecord(row), true, nil
}
// List returns every record for a project (no archive filter — mirrors the
// in-memory store contract; terminal filtering is the caller's job).
func (s *Store) List(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) {
rows, err := s.q.ListSessionsByProject(ctx, string(project))
if err != nil {
return nil, fmt.Errorf("list sessions for %s: %w", project, err)
}
out := make([]domain.SessionRecord, 0, len(rows))
for _, row := range rows {
out = append(out, rowToRecord(row))
}
return out, nil
}
// ListAll returns every persisted session across all projects. The CDC snapshot
// source uses it to rebuild current state after a log-rotation gap.
func (s *Store) ListAll(ctx context.Context) ([]domain.SessionRecord, error) {
rows, err := s.q.ListAllSessions(ctx)
if err != nil {
return nil, fmt.Errorf("list all sessions: %w", err)
}
out := make([]domain.SessionRecord, 0, len(rows))
for _, row := range rows {
out = append(out, rowToRecord(row))
}
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))
if err != nil {
return nil, 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
}
// 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 {
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()
}

View File

@ -0,0 +1,256 @@
package sqlite
import (
"context"
"strings"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
)
func newTestStore(t *testing.T) *Store {
t.Helper()
db, err := Open(t.TempDir())
if err != nil {
t.Fatalf("open: %v", err)
}
t.Cleanup(func() { db.Close() })
return NewStore(db)
}
func sampleRecord(id string) domain.SessionRecord {
now := time.Now().UTC().Truncate(time.Second)
return domain.SessionRecord{
ID: domain.SessionID(id),
ProjectID: "proj",
IssueID: "issue-1",
Kind: domain.KindWorker,
CreatedAt: now,
UpdatedAt: now,
Lifecycle: domain.CanonicalSessionLifecycle{
Session: domain.SessionSubstate{State: domain.SessionWorking, Reason: domain.ReasonTaskInProgress},
PR: domain.PRSubstate{State: domain.PRNone, Reason: domain.PRReasonNotCreated},
Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning},
Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: now, Source: domain.SourceNative},
},
}
}
func TestUpsertInsertThenUpdateBumpsRevision(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
rec := sampleRecord("s1")
if err := s.Upsert(ctx, rec, ports.EventSessionCreated); err != nil {
t.Fatalf("insert: %v", err)
}
lc, ok, err := s.Load(ctx, "s1")
if err != nil || !ok {
t.Fatalf("load after insert: ok=%v err=%v", ok, err)
}
if lc.Revision != 1 {
t.Fatalf("revision after insert = %d, want 1", lc.Revision)
}
// Update must carry the loaded revision (1) and persist as 2.
rec.Lifecycle.Revision = 1
rec.Lifecycle.Session.State = domain.SessionIdle
if err := s.Upsert(ctx, rec, ports.EventSessionStateChanged); err != nil {
t.Fatalf("update: %v", err)
}
lc, _, _ = s.Load(ctx, "s1")
if lc.Revision != 2 {
t.Fatalf("revision after update = %d, want 2", lc.Revision)
}
if lc.Session.State != domain.SessionIdle {
t.Fatalf("state after update = %q, want idle", lc.Session.State)
}
}
func TestUpsertStaleRevisionMismatch(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
rec := sampleRecord("s1")
if err := s.Upsert(ctx, rec, ports.EventSessionCreated); err != nil {
t.Fatalf("insert: %v", err)
}
// Stored revision is 1; submitting revision 0 (stale) must mismatch and
// write nothing new (no extra outbox/change_log rows).
rec.Lifecycle.Revision = 0
err := s.Upsert(ctx, rec, ports.EventSessionStateChanged)
if err == nil || !strings.Contains(err.Error(), "revision mismatch") {
t.Fatalf("stale update err = %v, want revision mismatch", err)
}
assertOutboxCount(t, s, ctx, 1)
}
func TestUpsertInsertNonZeroRevisionErrors(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
rec := sampleRecord("s1")
rec.Lifecycle.Revision = 5
err := s.Upsert(ctx, rec, ports.EventSessionCreated)
if err == nil || !strings.Contains(err.Error(), "revision mismatch") {
t.Fatalf("insert with revision 5 err = %v, want revision mismatch", err)
}
// Nothing should be persisted.
if _, ok, _ := s.Get(ctx, "s1"); ok {
t.Fatal("session persisted despite revision-mismatch insert")
}
assertOutboxCount(t, s, ctx, 0)
}
func TestUpsertOutboxAtomicityAndOrdering(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
rec := sampleRecord("s1")
if err := s.Upsert(ctx, rec, ports.EventSessionCreated); err != nil {
t.Fatalf("insert: %v", err)
}
rec.Lifecycle.Revision = 1
if err := s.Upsert(ctx, rec, ports.EventSessionStateChanged); err != nil {
t.Fatalf("update: %v", err)
}
rows, err := NewStore(s.db).q.ListUnsentOutbox(ctx, 100)
if err != nil {
t.Fatalf("list outbox: %v", err)
}
if len(rows) != 2 {
t.Fatalf("outbox rows = %d, want 2", len(rows))
}
// seq strictly monotonic, event types verbatim, revisions 1 then 2.
if rows[0].ChangeLogSeq != 1 || rows[1].ChangeLogSeq != 2 {
t.Fatalf("seq not monotonic: %d, %d", rows[0].ChangeLogSeq, rows[1].ChangeLogSeq)
}
if rows[0].EventType != string(ports.EventSessionCreated) || rows[1].EventType != string(ports.EventSessionStateChanged) {
t.Fatalf("event types = %q, %q", rows[0].EventType, rows[1].EventType)
}
if rows[0].Revision != 1 || rows[1].Revision != 2 {
t.Fatalf("revisions = %d, %d, want 1, 2", rows[0].Revision, rows[1].Revision)
}
}
func TestGetListRoundTrip(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
a := sampleRecord("a")
b := sampleRecord("b")
b.ProjectID = "other"
if err := s.Upsert(ctx, a, ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
if err := s.Upsert(ctx, b, ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
got, ok, err := s.Get(ctx, "a")
if err != nil || !ok {
t.Fatalf("get a: ok=%v err=%v", ok, err)
}
if got.ID != "a" || got.Lifecycle.Revision != 1 || got.IssueID != "issue-1" {
t.Fatalf("unexpected record: %+v", got)
}
if got.Metadata != nil {
t.Fatalf("Get must not reconstruct metadata, got %v", got.Metadata)
}
list, err := s.List(ctx, "proj")
if err != nil {
t.Fatal(err)
}
if len(list) != 1 || list[0].ID != "a" {
t.Fatalf("List(proj) = %+v, want only a", list)
}
}
func TestMetadataSideChannel(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
if err := s.Upsert(ctx, sampleRecord("s1"), ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
if err := s.PatchMetadata(ctx, "s1", map[string]string{"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 {
t.Fatalf("patch overwrite: %v", err)
}
m, err := s.GetMetadata(ctx, "s1")
if err != nil {
t.Fatal(err)
}
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")
if lc.Revision != 1 {
t.Fatalf("revision = %d after metadata patch, want 1 (no bump)", lc.Revision)
}
}
func TestDetectingRoundTrip(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
rec := sampleRecord("s1")
rec.Lifecycle.Session.State = domain.SessionDetecting
rec.Lifecycle.Detecting = &domain.DetectingState{
Attempts: 2,
StartedAt: time.Now().UTC().Truncate(time.Second),
EvidenceHash: "abc123",
}
if err := s.Upsert(ctx, rec, ports.EventSessionCreated); err != nil {
t.Fatal(err)
}
lc, _, _ := s.Load(ctx, "s1")
if lc.Detecting == nil {
t.Fatal("Detecting lost on round-trip")
}
if lc.Detecting.Attempts != 2 || lc.Detecting.EvidenceHash != "abc123" {
t.Fatalf("detecting = %+v", lc.Detecting)
}
// Clearing Detecting must null the columns back out.
rec.Lifecycle.Revision = 1
rec.Lifecycle.Detecting = nil
if err := s.Upsert(ctx, rec, ports.EventSessionStateChanged); err != nil {
t.Fatal(err)
}
lc, _, _ = s.Load(ctx, "s1")
if lc.Detecting != nil {
t.Fatalf("Detecting not cleared: %+v", lc.Detecting)
}
}
func TestLoadGetMissing(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
if _, ok, err := s.Load(ctx, "nope"); ok || err != nil {
t.Fatalf("Load missing: ok=%v err=%v", ok, err)
}
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 {
t.Fatalf("GetMetadata missing: m=%v err=%v", m, err)
}
}
func assertOutboxCount(t *testing.T, s *Store, ctx context.Context, want int) {
t.Helper()
rows, err := s.q.ListUnsentOutbox(ctx, 1000)
if err != nil {
t.Fatalf("list outbox: %v", err)
}
if len(rows) != want {
t.Fatalf("outbox count = %d, want %d", len(rows), want)
}
}

View File

@ -0,0 +1,113 @@
package sqlite
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
)
// Upsert performs the one atomic canonical write: it CAS-checks and persists the
// session row (bumping revision), appends a change_log entry, and enqueues an
// outbox row linked to that entry's seq — all in a single transaction. Only the
// LCM calls this.
//
// Revision CAS (mirrors the in-memory store contract exactly):
// - existing row: rec.Lifecycle.Revision must equal the stored revision, else
// a revision-mismatch error and nothing is written; on match it persists at
// stored+1.
// - insert: rec.Lifecycle.Revision must be 0, persisted as 1.
func (s *Store) Upsert(ctx context.Context, rec domain.SessionRecord, eventType ports.EventType) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("begin upsert: %w", err)
}
defer tx.Rollback()
qtx := s.q.WithTx(tx)
newRevision, err := casPersist(ctx, qtx, rec)
if err != nil {
return err
}
if err := appendOutbox(ctx, qtx, rec, newRevision, eventType); err != nil {
return err
}
return tx.Commit()
}
// casPersist applies the revision-CAS insert-or-update and returns the new
// stored revision.
func casPersist(ctx context.Context, q *gen.Queries, rec domain.SessionRecord) (int, error) {
stored, err := q.GetSessionRevision(ctx, string(rec.ID))
switch {
case errors.Is(err, sql.ErrNoRows):
// Insert path: incoming revision must be 0; row persists at revision 1.
if rec.Lifecycle.Revision != 0 {
return 0, fmt.Errorf("revision mismatch for insert %s: have %d, want 0", rec.ID, rec.Lifecycle.Revision)
}
rows, err := q.InsertSession(ctx, recordToInsert(rec))
if err != nil {
return 0, fmt.Errorf("insert session %s: %w", rec.ID, err)
}
if rows != 1 {
// Another writer raced us between the revision check and the insert.
// With single-writer this should not happen; treat as a CAS failure.
return 0, fmt.Errorf("revision mismatch for insert %s: row already exists", rec.ID)
}
return 1, nil
case err != nil:
return 0, fmt.Errorf("read revision %s: %w", rec.ID, err)
default:
// Update path: incoming revision must equal the stored revision.
if int64(rec.Lifecycle.Revision) != stored {
return 0, fmt.Errorf("revision mismatch for %s: have %d, want %d", rec.ID, rec.Lifecycle.Revision, stored)
}
rows, err := q.UpdateSessionCAS(ctx, recordToUpdate(rec, stored))
if err != nil {
return 0, fmt.Errorf("update session %s: %w", rec.ID, err)
}
if rows != 1 {
return 0, fmt.Errorf("revision mismatch for %s: stale revision %d", rec.ID, rec.Lifecycle.Revision)
}
return int(stored) + 1, nil
}
}
// 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).
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)
}
seq, err := q.InsertChangeLog(ctx, gen.InsertChangeLogParams{
SessionID: string(rec.ID),
EventType: string(eventType),
Revision: int64(newRevision),
Payload: string(blob),
CreatedAt: now,
})
if err != nil {
return fmt.Errorf("insert change_log %s: %w", rec.ID, err)
}
if err := q.InsertOutbox(ctx, gen.InsertOutboxParams{ChangeLogSeq: seq, CreatedAt: now}); err != nil {
return fmt.Errorf("insert outbox %s: %w", rec.ID, err)
}
return nil
}

126
backend/lifecycle_wiring.go Normal file
View File

@ -0,0 +1,126 @@
package main
import (
"context"
"log/slog"
"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"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
// lifecycleStack owns the running LCM + reaper. The LCM is the sole writer into
// the store (every Apply*/On* call ends in store.Upsert, which the CDC pipeline
// then drains); the reaper is the OBSERVE-layer timer that probes live runtimes
// and reports facts back through the LCM. Together with the CDC substrate this
// makes the write path live end-to-end: LCM -> store -> outbox -> JSONL ->
// broadcaster.
type lifecycleStack struct {
LCM *lifecycle.Manager
reaperDone <-chan struct{}
}
// startLifecycle constructs the LCM over store, makes escalation budgets durable,
// teaches it to enumerate sessions for the reaper, and starts the reaper loop.
// The goroutine stops when ctx is cancelled; Stop waits for it to drain.
//
// TEMPORARY STUBS (replace as the daemon lane lands the real collaborators):
//
// - noopNotifier — swap for the production notifier multiplexer once the
// notifier plugins (desktop/Slack/webhook) are ported. Wire it where
// noopNotifier{} is passed to lifecycle.New below.
// - noopMessenger — swap for the AgentMessenger backed by the runtime/agent
// plugins (it injects a prompt into the live agent pane). Wire it at the
// same lifecycle.New call site.
// - reaper.MapRegistry{} — empty runtime registry, so the reaper probes
// nothing. Register the real runtime adapters (tmux/process) keyed by
// runtime name once those plugins exist: reaper.MapRegistry{"tmux": rt}.
func startLifecycle(ctx context.Context, store *sqlite.Store, logger *slog.Logger) (*lifecycleStack, error) {
// TODO(daemon-lane): replace noopNotifier{}/noopMessenger{} with the real
// notifier multiplexer and the plugin-backed AgentMessenger.
lcm := lifecycle.New(store, noopNotifier{}, noopMessenger{})
// Durable escalation budgets (flaw #3 fix): hydrate from the store and turn
// on write-through so a restart does not re-fire an already-escalated page.
// Must run before the reaper starts dispatching TickEscalations.
if err := lcm.WithReactionStore(ctx, lifecycleReactionStore{store}); err != nil {
return nil, err
}
// The reaper's RunningSessions snapshot needs to see every session; ListAll
// spans all projects (the per-project List would hide cross-project work).
lcm.WithSessionLister(store.ListAll)
// TODO(daemon-lane): pass the real runtime registry so the reaper actually
// probes live panes. With an empty registry it ticks escalations but probes
// nothing, which is correct until runtimes exist.
rp := reaper.New(lcm, reaper.MapRegistry{}, reaper.Config{Logger: logger})
return &lifecycleStack{LCM: lcm, reaperDone: rp.Start(ctx)}, nil
}
// Stop waits for the reaper goroutine to exit (the caller must have cancelled the
// ctx passed to startLifecycle).
func (l *lifecycleStack) Stop() {
<-l.reaperDone
}
// noopNotifier satisfies ports.Notifier by dropping every event. TEMPORARY: the
// daemon lane replaces this with the notifier multiplexer over the real notifier
// plugins. Until then human-facing notifications are silently discarded — the
// write path and CDC still work, only the human push is absent.
type noopNotifier struct{}
func (noopNotifier) Notify(context.Context, ports.OrchestratorEvent) error { return nil }
// noopMessenger satisfies ports.AgentMessenger by dropping every send. TEMPORARY:
// replace with the runtime/agent-plugin-backed messenger that injects prompts
// into the live agent pane. Until then auto-nudge reactions are no-ops.
type noopMessenger struct{}
func (noopMessenger) Send(context.Context, domain.SessionID, string) error { return nil }
// lifecycleReactionStore bridges the concrete *sqlite.Store to the lifecycle
// package's ReactionStore interface (string/row types <-> domain types). It is
// the production twin of the reactionStoreAdapter used in the lifecycle tests.
type lifecycleReactionStore struct{ store *sqlite.Store }
func (a lifecycleReactionStore) LoadReactionTrackers(ctx context.Context) ([]lifecycle.PersistedTracker, error) {
rows, err := a.store.ListReactionTrackers(ctx)
if err != nil {
return nil, err
}
out := make([]lifecycle.PersistedTracker, len(rows))
for i, r := range rows {
out[i] = lifecycle.PersistedTracker{
SessionID: domain.SessionID(r.SessionID),
Key: r.ReactionKey,
Attempts: r.Attempts,
Escalated: r.Escalated,
FirstAttemptAt: r.FirstAttemptAt,
ProjectID: domain.ProjectID(r.ProjectID),
}
}
return out, nil
}
func (a lifecycleReactionStore) SaveReactionTracker(ctx context.Context, t lifecycle.PersistedTracker) error {
return a.store.SaveReactionTracker(ctx, sqlite.ReactionTrackerRow{
SessionID: string(t.SessionID),
ReactionKey: t.Key,
Attempts: t.Attempts,
Escalated: t.Escalated,
FirstAttemptAt: t.FirstAttemptAt,
ProjectID: string(t.ProjectID),
})
}
func (a lifecycleReactionStore) DeleteReactionTracker(ctx context.Context, id domain.SessionID, key string) error {
return a.store.DeleteReactionTracker(ctx, string(id), key)
}
func (a lifecycleReactionStore) DeleteSessionReactionTrackers(ctx context.Context, id domain.SessionID) error {
return a.store.DeleteSessionReactionTrackers(ctx, string(id))
}

View File

@ -15,6 +15,7 @@ import (
"github.com/aoagents/agent-orchestrator/backend/internal/config"
"github.com/aoagents/agent-orchestrator/backend/internal/httpd"
"github.com/aoagents/agent-orchestrator/backend/internal/runfile"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
func main() {
@ -46,11 +47,54 @@ func run() error {
return err
}
// Open the durable store and bring up the CDC substrate (outbox publisher,
// JSONL consumer + broadcaster, outbox janitor). The LCM/Session Manager and
// the HTTP API routes that drive and read this store are owned by the daemon
// lane and are wired there once their collaborators (Notifier, AgentMessenger,
// and the runtime/agent/workspace plugins) have production implementations;
// here we stand up the persistence + change-delivery foundation they build on.
db, err := sqlite.Open(cfg.DataDir)
if err != nil {
return fmt.Errorf("open store: %w", err)
}
defer db.Close()
store := sqlite.NewStore(db)
// signal.NotifyContext cancels ctx on SIGINT/SIGTERM, which drives the
// graceful shutdown inside Server.Run.
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
cdcPipe, err := startCDC(ctx, store, cfg.DataDir, log)
if err != nil {
return err
}
defer func() {
if err := cdcPipe.Stop(); err != nil {
log.Error("cdc pipeline shutdown", "err", err)
}
}()
// Bring up the Lifecycle Manager (sole store writer) and the reaper (OBSERVE
// timer). This makes the write path live end-to-end: LCM.Upsert -> store ->
// outbox -> CDC JSONL -> broadcaster. The collaborators it needs that don't
// yet have production implementations (Notifier, AgentMessenger, runtime
// registry) are stubbed in lifecycle_wiring.go with TODO markers.
//
// NOT wired here yet — both await collaborators the daemon lane owns:
// - Session Manager: session.New needs Runtime/Agent/Workspace plugins to
// construct. Stubbing them would make Spawn a silent no-op (a footgun),
// so it's deferred rather than faked. The LCM already exposes the read
// surface (RunningSessions) the SM would wrap.
// - HTTP API routes: httpd.New takes no SM/LCM today; surfacing the store
// over HTTP needs a constructor signature change + handlers, tracked with
// the SM work since the routes call into it.
lcStack, err := startLifecycle(ctx, store, log)
if err != nil {
return fmt.Errorf("start lifecycle: %w", err)
}
defer lcStack.Stop()
return srv.Run(ctx)
}

134
backend/main_test.go Normal file
View File

@ -0,0 +1,134 @@
package main
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite"
)
// These tests cover the composition-root adapters in cdc_wiring.go directly
// (package main otherwise has no test coverage): the outboxAdapter mapping the
// store's OutboxEvent to cdc.PendingEvent, and the snapshotSource rebuilding
// full-state events from the sessions table.
func newWiringStore(t *testing.T) *sqlite.Store {
t.Helper()
db, err := sqlite.Open(t.TempDir())
if err != nil {
t.Fatalf("open: %v", err)
}
t.Cleanup(func() { db.Close() })
return sqlite.NewStore(db)
}
func wiringRec(id string) domain.SessionRecord {
now := time.Now().UTC()
return domain.SessionRecord{
ID: domain.SessionID(id), ProjectID: "proj", Kind: domain.KindWorker, CreatedAt: now, UpdatedAt: now,
Lifecycle: domain.CanonicalSessionLifecycle{
Session: domain.SessionSubstate{State: domain.SessionWorking, Reason: domain.ReasonTaskInProgress},
PR: domain.PRSubstate{State: domain.PRNone, Reason: domain.PRReasonNotCreated},
Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning},
Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: now, Source: domain.SourceNative},
},
}
}
func TestOutboxAdapterMapsPendingEvents(t *testing.T) {
ctx := context.Background()
store := newWiringStore(t)
a := outboxAdapter{store}
if err := store.Upsert(ctx, wiringRec("s1"), ports.EventSessionCreated); err != nil {
t.Fatalf("upsert: %v", err)
}
pending, err := a.ListUnsent(ctx, 10)
if err != nil {
t.Fatalf("list unsent: %v", err)
}
if len(pending) != 1 {
t.Fatalf("want 1 pending event, got %d", len(pending))
}
pe := pending[0]
if pe.Seq != 1 || pe.SessionID != "s1" || pe.EventType != string(ports.EventSessionCreated) || pe.Revision != 1 {
t.Fatalf("unexpected mapping: %+v", pe)
}
if pe.Payload == "" {
t.Fatal("payload should carry the marshaled record")
}
// MarkSent must clear it from the unsent set.
if err := a.MarkSent(ctx, pe.OutboxID, time.Now().UTC()); err != nil {
t.Fatalf("mark sent: %v", err)
}
again, err := a.ListUnsent(ctx, 10)
if err != nil {
t.Fatalf("list unsent 2: %v", err)
}
if len(again) != 0 {
t.Fatalf("sent event should not reappear, got %d", len(again))
}
}
func TestSnapshotSourceRebuildsState(t *testing.T) {
ctx := context.Background()
store := newWiringStore(t)
s := snapshotSource{store}
// Empty store: no events, maxSeq 0.
events, maxSeq, err := s.Snapshot(ctx)
if err != nil {
t.Fatalf("empty snapshot: %v", err)
}
if len(events) != 0 || maxSeq != 0 {
t.Fatalf("empty store should yield no events and maxSeq 0, got %d events maxSeq %d", len(events), maxSeq)
}
// Two canonical writes (seq 1,2) across two sessions.
if err := store.Upsert(ctx, wiringRec("s1"), ports.EventSessionCreated); err != nil {
t.Fatalf("upsert s1: %v", err)
}
if err := store.Upsert(ctx, wiringRec("s2"), ports.EventSessionCreated); err != nil {
t.Fatalf("upsert s2: %v", err)
}
events, maxSeq, err = s.Snapshot(ctx)
if err != nil {
t.Fatalf("snapshot: %v", err)
}
if maxSeq != 2 {
t.Fatalf("maxSeq = %d, want 2 (change_log high-water)", maxSeq)
}
if len(events) != 2 {
t.Fatalf("want one event per session (2), got %d", len(events))
}
for _, e := range events {
if e.Seq != maxSeq {
t.Errorf("snapshot event seq = %d, want resume watermark %d", e.Seq, maxSeq)
}
if e.EventType != "session_snapshot" {
t.Errorf("event type = %q, want session_snapshot", e.EventType)
}
// Payload must be a parseable full record at the persisted revision with
// metadata excluded and the schema version stamped.
var rec domain.SessionRecord
if err := json.Unmarshal([]byte(e.Payload), &rec); err != nil {
t.Fatalf("payload not a SessionRecord: %v", err)
}
if rec.Lifecycle.Version != domain.LifecycleVersion {
t.Errorf("payload version = %d, want %d", rec.Lifecycle.Version, domain.LifecycleVersion)
}
if rec.Lifecycle.Revision != 1 {
t.Errorf("payload revision = %d, want 1", rec.Lifecycle.Revision)
}
if rec.Metadata != nil {
t.Errorf("snapshot payload must exclude metadata, got %v", rec.Metadata)
}
}
}

13
backend/sqlc.yaml Normal file
View File

@ -0,0 +1,13 @@
version: "2"
sql:
- engine: "sqlite"
schema: "internal/storage/sqlite/migrations"
queries: "internal/storage/sqlite/queries"
gen:
go:
package: "gen"
out: "internal/storage/sqlite/gen"
emit_json_tags: false
emit_prepared_queries: false
emit_interface: true
emit_empty_slices: true