From f5bc4c7b8c70ff1bf9577bc8611beb016af174c0 Mon Sep 17 00:00:00 2001 From: Pritom14 Date: Sat, 30 May 2026 16:02:07 +0530 Subject: [PATCH] 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. --- .gitignore | 9 + backend/cdc_wiring.go | 143 ++++++++ backend/go.mod | 24 +- backend/go.sum | 66 ++++ backend/internal/cdc/broadcast.go | 44 +++ backend/internal/cdc/cdc_integration_test.go | 256 +++++++++++++++ backend/internal/cdc/consumer.go | 221 +++++++++++++ backend/internal/cdc/event.go | 32 ++ backend/internal/cdc/janitor.go | 84 +++++ backend/internal/cdc/jsonl.go | 109 +++++++ backend/internal/cdc/publisher.go | 115 +++++++ backend/internal/config/config.go | 24 ++ backend/internal/lifecycle/manager.go | 7 +- .../internal/lifecycle/manager_parity_test.go | 144 ++++++++ .../lifecycle/reaction_durability_test.go | 140 ++++++++ backend/internal/lifecycle/reaction_store.go | 94 ++++++ backend/internal/lifecycle/reactions.go | 30 +- backend/internal/storage/sqlite/cdc_store.go | 104 ++++++ backend/internal/storage/sqlite/db.go | 63 ++++ .../internal/storage/sqlite/gen/cdc.sql.go | 199 ++++++++++++ backend/internal/storage/sqlite/gen/db.go | 31 ++ .../storage/sqlite/gen/metadata.sql.go | 59 ++++ backend/internal/storage/sqlite/gen/models.go | 74 +++++ .../internal/storage/sqlite/gen/querier.go | 42 +++ .../storage/sqlite/gen/reactions.sql.go | 100 ++++++ .../storage/sqlite/gen/sessions.sql.go | 307 ++++++++++++++++++ backend/internal/storage/sqlite/mapping.go | 129 ++++++++ .../storage/sqlite/migrations/0001_init.sql | 109 +++++++ .../internal/storage/sqlite/queries/cdc.sql | 42 +++ .../storage/sqlite/queries/metadata.sql | 7 + .../storage/sqlite/queries/reactions.sql | 18 + .../storage/sqlite/queries/sessions.sql | 58 ++++ .../internal/storage/sqlite/reaction_store.go | 80 +++++ backend/internal/storage/sqlite/spike_test.go | 92 ++++++ backend/internal/storage/sqlite/store.go | 118 +++++++ backend/internal/storage/sqlite/store_test.go | 256 +++++++++++++++ backend/internal/storage/sqlite/upsert.go | 113 +++++++ backend/lifecycle_wiring.go | 126 +++++++ backend/main.go | 44 +++ backend/main_test.go | 134 ++++++++ backend/sqlc.yaml | 13 + 41 files changed, 3849 insertions(+), 11 deletions(-) create mode 100644 backend/cdc_wiring.go create mode 100644 backend/internal/cdc/broadcast.go create mode 100644 backend/internal/cdc/cdc_integration_test.go create mode 100644 backend/internal/cdc/consumer.go create mode 100644 backend/internal/cdc/event.go create mode 100644 backend/internal/cdc/janitor.go create mode 100644 backend/internal/cdc/jsonl.go create mode 100644 backend/internal/cdc/publisher.go create mode 100644 backend/internal/lifecycle/manager_parity_test.go create mode 100644 backend/internal/lifecycle/reaction_durability_test.go create mode 100644 backend/internal/lifecycle/reaction_store.go create mode 100644 backend/internal/storage/sqlite/cdc_store.go create mode 100644 backend/internal/storage/sqlite/db.go create mode 100644 backend/internal/storage/sqlite/gen/cdc.sql.go create mode 100644 backend/internal/storage/sqlite/gen/db.go create mode 100644 backend/internal/storage/sqlite/gen/metadata.sql.go create mode 100644 backend/internal/storage/sqlite/gen/models.go create mode 100644 backend/internal/storage/sqlite/gen/querier.go create mode 100644 backend/internal/storage/sqlite/gen/reactions.sql.go create mode 100644 backend/internal/storage/sqlite/gen/sessions.sql.go create mode 100644 backend/internal/storage/sqlite/mapping.go create mode 100644 backend/internal/storage/sqlite/migrations/0001_init.sql create mode 100644 backend/internal/storage/sqlite/queries/cdc.sql create mode 100644 backend/internal/storage/sqlite/queries/metadata.sql create mode 100644 backend/internal/storage/sqlite/queries/reactions.sql create mode 100644 backend/internal/storage/sqlite/queries/sessions.sql create mode 100644 backend/internal/storage/sqlite/reaction_store.go create mode 100644 backend/internal/storage/sqlite/spike_test.go create mode 100644 backend/internal/storage/sqlite/store.go create mode 100644 backend/internal/storage/sqlite/store_test.go create mode 100644 backend/internal/storage/sqlite/upsert.go create mode 100644 backend/lifecycle_wiring.go create mode 100644 backend/main_test.go create mode 100644 backend/sqlc.yaml diff --git a/.gitignore b/.gitignore index e5ea212a2..425b31d78 100644 --- a/.gitignore +++ b/.gitignore @@ -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.* diff --git a/backend/cdc_wiring.go b/backend/cdc_wiring.go new file mode 100644 index 000000000..89997e7dc --- /dev/null +++ b/backend/cdc_wiring.go @@ -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 +} diff --git a/backend/go.mod b/backend/go.mod index 311cea288..88ca590cf 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -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 +) diff --git a/backend/go.sum b/backend/go.sum index 823cdbb1a..89f839295 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -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= diff --git a/backend/internal/cdc/broadcast.go b/backend/internal/cdc/broadcast.go new file mode 100644 index 000000000..a7458e38d --- /dev/null +++ b/backend/internal/cdc/broadcast.go @@ -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) + } +} diff --git a/backend/internal/cdc/cdc_integration_test.go b/backend/internal/cdc/cdc_integration_test.go new file mode 100644 index 000000000..9390afe01 --- /dev/null +++ b/backend/internal/cdc/cdc_integration_test.go @@ -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) + } +} diff --git a/backend/internal/cdc/consumer.go b/backend/internal/cdc/consumer.go new file mode 100644 index 000000000..00edb0f10 --- /dev/null +++ b/backend/internal/cdc/consumer.go @@ -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" +} diff --git a/backend/internal/cdc/event.go b/backend/internal/cdc/event.go new file mode 100644 index 000000000..b0eddf982 --- /dev/null +++ b/backend/internal/cdc/event.go @@ -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"` +} diff --git a/backend/internal/cdc/janitor.go b/backend/internal/cdc/janitor.go new file mode 100644 index 000000000..3968b2cf4 --- /dev/null +++ b/backend/internal/cdc/janitor.go @@ -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) +} diff --git a/backend/internal/cdc/jsonl.go b/backend/internal/cdc/jsonl.go new file mode 100644 index 000000000..74dc0695c --- /dev/null +++ b/backend/internal/cdc/jsonl.go @@ -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() +} diff --git a/backend/internal/cdc/publisher.go b/backend/internal/cdc/publisher.go new file mode 100644 index 000000000..3283a236e --- /dev/null +++ b/backend/internal/cdc/publisher.go @@ -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 +} diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index d6765dba2..68aab00e7 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -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 /running.json) +// AO_DATA_DIR durable state dir (default /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 +} diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index b5751e867..54e6887f0 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -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 } diff --git a/backend/internal/lifecycle/manager_parity_test.go b/backend/internal/lifecycle/manager_parity_test.go new file mode 100644 index 000000000..146dcc16c --- /dev/null +++ b/backend/internal/lifecycle/manager_parity_test.go @@ -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) + } + } +} diff --git a/backend/internal/lifecycle/reaction_durability_test.go b/backend/internal/lifecycle/reaction_durability_test.go new file mode 100644 index 000000000..1866c8c97 --- /dev/null +++ b/backend/internal/lifecycle/reaction_durability_test.go @@ -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)) + } +} diff --git a/backend/internal/lifecycle/reaction_store.go b/backend/internal/lifecycle/reaction_store.go new file mode 100644 index 000000000..f8da7415b --- /dev/null +++ b/backend/internal/lifecycle/reaction_store.go @@ -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) +} diff --git a/backend/internal/lifecycle/reactions.go b/backend/internal/lifecycle/reactions.go index 26dea5627..ac4de400a 100644 --- a/backend/internal/lifecycle/reactions.go +++ b/backend/internal/lifecycle/reactions.go @@ -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 } diff --git a/backend/internal/storage/sqlite/cdc_store.go b/backend/internal/storage/sqlite/cdc_store.go new file mode 100644 index 000000000..3386f9880 --- /dev/null +++ b/backend/internal/storage/sqlite/cdc_store.go @@ -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) +} diff --git a/backend/internal/storage/sqlite/db.go b/backend/internal/storage/sqlite/db.go new file mode 100644 index 000000000..78eb3ae9d --- /dev/null +++ b/backend/internal/storage/sqlite/db.go @@ -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 +} diff --git a/backend/internal/storage/sqlite/gen/cdc.sql.go b/backend/internal/storage/sqlite/gen/cdc.sql.go new file mode 100644 index 000000000..c2eedc8c7 --- /dev/null +++ b/backend/internal/storage/sqlite/gen/cdc.sql.go @@ -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 +} diff --git a/backend/internal/storage/sqlite/gen/db.go b/backend/internal/storage/sqlite/gen/db.go new file mode 100644 index 000000000..b6fcf6be3 --- /dev/null +++ b/backend/internal/storage/sqlite/gen/db.go @@ -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, + } +} diff --git a/backend/internal/storage/sqlite/gen/metadata.sql.go b/backend/internal/storage/sqlite/gen/metadata.sql.go new file mode 100644 index 000000000..96510eb80 --- /dev/null +++ b/backend/internal/storage/sqlite/gen/metadata.sql.go @@ -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 +} diff --git a/backend/internal/storage/sqlite/gen/models.go b/backend/internal/storage/sqlite/gen/models.go new file mode 100644 index 000000000..210fe245a --- /dev/null +++ b/backend/internal/storage/sqlite/gen/models.go @@ -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 +} diff --git a/backend/internal/storage/sqlite/gen/querier.go b/backend/internal/storage/sqlite/gen/querier.go new file mode 100644 index 000000000..074fe053c --- /dev/null +++ b/backend/internal/storage/sqlite/gen/querier.go @@ -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) diff --git a/backend/internal/storage/sqlite/gen/reactions.sql.go b/backend/internal/storage/sqlite/gen/reactions.sql.go new file mode 100644 index 000000000..dc7b01c2a --- /dev/null +++ b/backend/internal/storage/sqlite/gen/reactions.sql.go @@ -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 +} diff --git a/backend/internal/storage/sqlite/gen/sessions.sql.go b/backend/internal/storage/sqlite/gen/sessions.sql.go new file mode 100644 index 000000000..00d97ad66 --- /dev/null +++ b/backend/internal/storage/sqlite/gen/sessions.sql.go @@ -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() +} diff --git a/backend/internal/storage/sqlite/mapping.go b/backend/internal/storage/sqlite/mapping.go new file mode 100644 index 000000000..39ae21275 --- /dev/null +++ b/backend/internal/storage/sqlite/mapping.go @@ -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, + } +} diff --git a/backend/internal/storage/sqlite/migrations/0001_init.sql b/backend/internal/storage/sqlite/migrations/0001_init.sql new file mode 100644 index 000000000..f343e16d1 --- /dev/null +++ b/backend/internal/storage/sqlite/migrations/0001_init.sql @@ -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 diff --git a/backend/internal/storage/sqlite/queries/cdc.sql b/backend/internal/storage/sqlite/queries/cdc.sql new file mode 100644 index 000000000..b818194a5 --- /dev/null +++ b/backend/internal/storage/sqlite/queries/cdc.sql @@ -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 < ?; diff --git a/backend/internal/storage/sqlite/queries/metadata.sql b/backend/internal/storage/sqlite/queries/metadata.sql new file mode 100644 index 000000000..45079bb25 --- /dev/null +++ b/backend/internal/storage/sqlite/queries/metadata.sql @@ -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; diff --git a/backend/internal/storage/sqlite/queries/reactions.sql b/backend/internal/storage/sqlite/queries/reactions.sql new file mode 100644 index 000000000..0ccd99c3f --- /dev/null +++ b/backend/internal/storage/sqlite/queries/reactions.sql @@ -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 = ?; diff --git a/backend/internal/storage/sqlite/queries/sessions.sql b/backend/internal/storage/sqlite/queries/sessions.sql new file mode 100644 index 000000000..48cdcacf1 --- /dev/null +++ b/backend/internal/storage/sqlite/queries/sessions.sql @@ -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; diff --git a/backend/internal/storage/sqlite/reaction_store.go b/backend/internal/storage/sqlite/reaction_store.go new file mode 100644 index 000000000..819d9716b --- /dev/null +++ b/backend/internal/storage/sqlite/reaction_store.go @@ -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) +} diff --git a/backend/internal/storage/sqlite/spike_test.go b/backend/internal/storage/sqlite/spike_test.go new file mode 100644 index 000000000..30b43fc7c --- /dev/null +++ b/backend/internal/storage/sqlite/spike_test.go @@ -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]) + } +} diff --git a/backend/internal/storage/sqlite/store.go b/backend/internal/storage/sqlite/store.go new file mode 100644 index 000000000..bd61e73b3 --- /dev/null +++ b/backend/internal/storage/sqlite/store.go @@ -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() +} diff --git a/backend/internal/storage/sqlite/store_test.go b/backend/internal/storage/sqlite/store_test.go new file mode 100644 index 000000000..5457855da --- /dev/null +++ b/backend/internal/storage/sqlite/store_test.go @@ -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) + } +} diff --git a/backend/internal/storage/sqlite/upsert.go b/backend/internal/storage/sqlite/upsert.go new file mode 100644 index 000000000..40944005a --- /dev/null +++ b/backend/internal/storage/sqlite/upsert.go @@ -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 +} diff --git a/backend/lifecycle_wiring.go b/backend/lifecycle_wiring.go new file mode 100644 index 000000000..3836baf68 --- /dev/null +++ b/backend/lifecycle_wiring.go @@ -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)) +} diff --git a/backend/main.go b/backend/main.go index 78a232927..8db058ea5 100644 --- a/backend/main.go +++ b/backend/main.go @@ -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) } diff --git a/backend/main_test.go b/backend/main_test.go new file mode 100644 index 000000000..1a8d60c3f --- /dev/null +++ b/backend/main_test.go @@ -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) + } + } +} diff --git a/backend/sqlc.yaml b/backend/sqlc.yaml new file mode 100644 index 000000000..9659bf779 --- /dev/null +++ b/backend/sqlc.yaml @@ -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