diff --git a/internal/connector/admission/commit.go b/internal/connector/admission/commit.go
index f8d356055..13a3cbce0 100644
--- a/internal/connector/admission/commit.go
+++ b/internal/connector/admission/commit.go
@@ -13,7 +13,7 @@ import (
var ErrAlreadyDecided = errors.New("admission: event already decided")
// Ledger is the durable half of a verdict. Intake's SQLite ledger implements
-// it once basecamp-cli PR 729 merges; until then only tests do.
+// it (internal/connector, Ledger.Admission).
type Ledger interface {
// Commit writes v onto its event's record in one transaction and returns
// the state it wrote. Within that transaction it must:
diff --git a/internal/connector/admission/doc.go b/internal/connector/admission/doc.go
index 1167b0fb0..0e6626566 100644
--- a/internal/connector/admission/doc.go
+++ b/internal/connector/admission/doc.go
@@ -73,13 +73,14 @@
//
// # The seam with intake
//
-// Intake (basecamp-cli PR 729, internal/connector) hands admission an event id
-// through its Queue; the record behind that id is the ledger row, whose
-// pointer fields are exactly Event's. Run is written against two small
-// interfaces, IDSource (satisfied by that Queue's Take) and Records (a ledger
-// read returning a seen record as an Event), plus Ledger, whose Commit
-// contract is invariant 5. The adapter from intake's ledger to those
-// interfaces lands once 729 merges; this package imports nothing from it.
+// Intake (internal/connector) hands admission an event id through its Queue;
+// the record behind that id is the ledger row, whose pointer fields are
+// exactly Event's. Run is written against two small interfaces, IDSource
+// (satisfied by that Queue's Take) and Records (a ledger read returning a seen
+// record as an Event), plus Ledger, whose Commit contract is invariant 5.
+// Intake's ledger implements Records and Ledger (connector.Ledger.Admission),
+// and connector.RunAdmission wires Run onto the queue; this package imports
+// nothing from that one.
//
// # What is not here
//
diff --git a/internal/connector/admission/event.go b/internal/connector/admission/event.go
index 1a19431ff..bcd23a0f6 100644
--- a/internal/connector/admission/event.go
+++ b/internal/connector/admission/event.go
@@ -10,7 +10,7 @@ import (
const ActorTypeAgent = "agent"
// Event is the pointer admission decides on: the fields intake stores for a
-// seen record (internal/connector Record in basecamp-cli PR 729), and nothing
+// seen record (internal/connector's Record), and nothing
// else. It carries no title, no content, no URL and no names; whatever
// admission needs beyond ids it reads.
type Event struct {
diff --git a/internal/connector/admission/run.go b/internal/connector/admission/run.go
index 421d94f02..37fcf079b 100644
--- a/internal/connector/admission/run.go
+++ b/internal/connector/admission/run.go
@@ -15,8 +15,8 @@ import (
// DefaultWorkers is the admission fetcher pool size.
const DefaultWorkers = 4
-// IDSource hands over the next event id to admit. Intake's Queue (basecamp-cli
-// PR 729, internal/connector) satisfies it.
+// IDSource hands over the next event id to admit. Intake's Queue
+// (internal/connector) satisfies it.
type IDSource interface {
Take(ctx context.Context) (int64, error)
}
@@ -27,8 +27,8 @@ type IDSource interface {
// false when the id is unknown or the record is past deciding, so it is
// skipped rather than decided twice.
//
-// This is the seam where intake hands admission an event: the adapter over
-// intake's Ledger.Get lands once PR 729 merges.
+// This is the seam where intake hands admission an event; intake's ledger
+// implements it (internal/connector, Ledger.Admission).
type Records interface {
LoadUndecided(ctx context.Context, id int64) (ev Event, ok bool, err error)
}
diff --git a/internal/connector/intake.go b/internal/connector/intake.go
index e47d2e5b3..21708f154 100644
--- a/internal/connector/intake.go
+++ b/internal/connector/intake.go
@@ -855,9 +855,9 @@ func (in *Intake) sweepStranded(ctx context.Context) {
// so that a crash can never lose the pointer — but that ordering means a
// failure after the commit leaves an event the ledger's own dedupe will
// suppress on every retry. Nothing may be left in that state, so a start
-// offers every record nothing has judged yet. Offering one twice costs
-// nothing: the queue carries ids, and admission moves a record out of seen
-// before it acts.
+// offers every record nothing has judged yet. Offering one twice costs at
+// most a second decision, never a second verdict: the queue carries ids, and
+// admission's commit applies only at the revision its decision loaded.
//
// The pointer line is NOT written again here. A record that got no line when
// it was recorded — the write that failed — never gets one, and the work still
diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go
index 0a9589bd3..90a4fdfe6 100644
--- a/internal/connector/ledger.go
+++ b/internal/connector/ledger.go
@@ -281,6 +281,47 @@ WHEN OLD.state IN ('completed', 'discarded') AND NEW.state <> OLD.state
BEGIN
SELECT RAISE(ABORT, 'a terminal record cannot change state');
END;
+`,
+ // Migration 4. What admission decides is written onto the record it
+ // decided, in the transaction that decides it.
+ //
+ // revision is the guard on that write: a decision carries the revision
+ // the record was loaded at, and applies only while the record is still
+ // there, so one event gets one verdict however many fetches decide it and
+ // an older decision never overwrites a newer one. Every state change bumps
+ // it, not only admission's.
+ //
+ // decided_at is when the latest verdict was written, blocked_at when the
+ // record entered its current run of blocked verdicts (the retry window
+ // counts from it), and retry_at a throttled verdict's server deadline.
+ // The rest is the verdict itself — what dispatch starts a task from, and
+ // what a blocked(no_route) record's holding reply needs. snapshot is the
+ // recording's content as admission read it. Only an admitted verdict
+ // writes one, whether the ledger writes it as admitted or as queued; a
+ // record keeps it through dispatch and completion until retention drops
+ // it, and loses it on any move to blocked or discarded. Nothing else in
+ // the ledger is content.
+ //
+ // Nothing has shipped that wrote a version 3 ledger, but a migration is
+ // how the schema changes regardless: the next time something has, this is
+ // the path that has to work.
+ `
+ALTER TABLE events ADD COLUMN revision INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE events ADD COLUMN decided_at TEXT;
+ALTER TABLE events ADD COLUMN blocked_at TEXT;
+ALTER TABLE events ADD COLUMN retry_at TEXT;
+ALTER TABLE events ADD COLUMN trigger_name TEXT NOT NULL DEFAULT '';
+ALTER TABLE events ADD COLUMN acknowledge INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE events ADD COLUMN conversation_key TEXT NOT NULL DEFAULT '';
+ALTER TABLE events ADD COLUMN reply_kind TEXT NOT NULL DEFAULT '';
+ALTER TABLE events ADD COLUMN reply_recording_id INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE events ADD COLUMN routed INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE events ADD COLUMN route TEXT NOT NULL DEFAULT '';
+ALTER TABLE events ADD COLUMN class TEXT NOT NULL DEFAULT '';
+ALTER TABLE events ADD COLUMN recording_url TEXT NOT NULL DEFAULT '';
+ALTER TABLE events ADD COLUMN requester_id INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE events ADD COLUMN snapshot BLOB;
+CREATE INDEX events_conversation ON events (conversation_key, state);
`,
}
diff --git a/internal/connector/ledger_admission.go b/internal/connector/ledger_admission.go
new file mode 100644
index 000000000..d46aad1d2
--- /dev/null
+++ b/internal/connector/ledger_admission.go
@@ -0,0 +1,250 @@
+package connector
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log/slog"
+ "time"
+
+ "github.com/basecamp/basecamp-cli/internal/connector/admission"
+ "github.com/basecamp/basecamp-cli/internal/connector/ndjson"
+)
+
+// Admission is the ledger as admission sees it: the Records it loads events
+// from and the Ledger its verdicts are committed to.
+//
+// Admission is its own package and imports nothing from this one. This is the
+// whole of the join: the pointer intake recorded goes out as an admission
+// Event with the revision it was loaded at, and a verdict comes back through
+// the ledger's one state-changing write, so the lifecycle that keeps a
+// finished record finished is the same lifecycle a verdict meets.
+type Admission struct {
+ ledger *Ledger
+}
+
+var (
+ _ admission.Records = Admission{}
+ _ admission.Ledger = Admission{}
+)
+
+// Admission returns the ledger's admission seam.
+func (l *Ledger) Admission() Admission { return Admission{ledger: l} }
+
+// undecided is where a record may be for admission to decide it: seen, which
+// nothing has judged yet, and blocked, which recovery and redispatch decide
+// again. Everything past these was decided by admission or moved on by
+// dispatch, and is never decided a second time.
+var undecided = []RecordState{StateSeen, StateBlocked}
+
+// LoadUndecided loads a seen or blocked record as the event admission decides,
+// with the revision it was loaded at. An unknown id, or a record past
+// deciding, is not ok and is skipped.
+func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event, bool, error) {
+ record, ok, err := a.ledger.Get(ctx, id)
+ if err != nil || !ok {
+ return admission.Event{}, false, err
+ }
+ if record.State != StateSeen && record.State != StateBlocked {
+ return admission.Event{}, false, nil
+ }
+ return admission.Event{
+ ID: record.ID,
+ EventType: record.EventType,
+ BucketID: record.BucketID,
+ RecordingID: record.RecordingID,
+ CreatorID: record.CreatorID,
+ PerformedByID: record.PerformedByID,
+ ActorType: record.ActorType,
+ Details: record.Details,
+ Revision: record.Revision,
+ SeenAt: record.SeenAt,
+ }, true, nil
+}
+
+// Commit writes a verdict onto its record in one transaction, and returns the
+// state it wrote. It holds admission.Ledger's contract:
+//
+// - the write applies only while the record is still at the verdict's
+// revision and still undecided, and it bumps the revision; otherwise it
+// is admission.ErrAlreadyDecided;
+// - an admitted verdict is written as queued when its conversation is
+// live, decided inside the transaction: another record on the key is
+// admitted and not yet dispatched (it becomes the task) or dispatched
+// (the task is running). A queued record alone is not a live
+// conversation: queued records join their task before it closes, and
+// counting one left behind would queue every later event on the key
+// with nothing to start a task;
+// - content is written only with an admitted verdict;
+// - a throttled verdict keeps its RetryAt.
+func (a Admission) Commit(ctx context.Context, v admission.Verdict) (admission.State, error) {
+ state, err := verdictState(v)
+ if err != nil {
+ return "", err
+ }
+ var snapshot []byte
+ if v.Snapshot != nil {
+ if snapshot, err = json.Marshal(v.Snapshot); err != nil {
+ return "", fmt.Errorf("connector: commit verdict on %d: %w", v.EventID, err)
+ }
+ }
+
+ var written RecordState
+ err = retryBusy(func() error {
+ var err error
+ written, err = a.commit(ctx, v, state, snapshot)
+ return err
+ })
+ if err != nil {
+ return "", err
+ }
+ return admission.State(written), nil
+}
+
+func (a Admission) commit(ctx context.Context, v admission.Verdict, state RecordState, snapshot []byte) (RecordState, error) {
+ l := a.ledger
+ tx, err := l.db.BeginTx(ctx, nil)
+ if err != nil {
+ return "", fmt.Errorf("connector: begin verdict on %d: %w", v.EventID, err)
+ }
+ defer func() { _ = tx.Rollback() }()
+
+ if state == StateAdmitted {
+ // The record being decided is seen or blocked, so it never counts
+ // itself as the conversation's task.
+ var live bool
+ if err := tx.QueryRowContext(ctx, liveConversation, v.ConversationKey, string(StateAdmitted), string(StateDispatched)).Scan(&live); err != nil {
+ return "", fmt.Errorf("connector: read conversation of %d: %w", v.EventID, err)
+ }
+ if live {
+ state = StateQueued
+ }
+ }
+
+ var retryAt time.Time
+ if state == StateBlocked {
+ retryAt = v.RetryAt
+ }
+ reply := admission.ReplyDestination{}
+ if v.Reply != nil {
+ reply = *v.Reply
+ }
+ revision := v.Revision
+ moved, err := l.move(ctx, tx, transition{
+ id: v.EventID,
+ state: state,
+ reason: string(v.Reason),
+ from: undecided,
+ revision: &revision,
+ retryAt: retryAt,
+ set: []assignment{
+ {column: "decided_at", value: l.timestamp()},
+ {column: "trigger_name", value: string(v.Trigger)},
+ {column: "acknowledge", value: v.Acknowledge},
+ {column: "conversation_key", value: v.ConversationKey},
+ {column: "reply_kind", value: string(reply.Kind)},
+ {column: "reply_recording_id", value: reply.RecordingID},
+ {column: "routed", value: v.Routed},
+ {column: "route", value: v.Route},
+ {column: "class", value: v.Class},
+ {column: "recording_url", value: v.RecordingURL},
+ {column: "requester_id", value: v.RequesterID},
+ {column: "snapshot", value: snapshot},
+ },
+ })
+ if err != nil {
+ return "", err
+ }
+ if !moved {
+ return "", explainVerdictRefusal(ctx, tx, v)
+ }
+ if err := tx.Commit(); err != nil {
+ return "", fmt.Errorf("connector: commit verdict on %d: %w", v.EventID, err)
+ }
+ return state, nil
+}
+
+// liveConversation asks whether a conversation has a task running or a record
+// about to become one. It is read through events_conversation.
+const liveConversation = `
+SELECT EXISTS (
+ SELECT 1 FROM events
+ WHERE conversation_key = ? AND state IN (?, ?)
+)`
+
+// verdictState checks a verdict is one the ledger can write and maps it to
+// the ledger's state. Queued is the ledger's decision, never a verdict's.
+func verdictState(v admission.Verdict) (RecordState, error) {
+ refuse := func(why string) (RecordState, error) {
+ return "", fmt.Errorf("connector: verdict on %d refused: %s", v.EventID, why)
+ }
+ switch v.State {
+ case admission.StateAdmitted:
+ if v.Snapshot == nil {
+ return refuse("an admitted verdict carries the recording it admitted")
+ }
+ if v.ConversationKey == "" {
+ return refuse("an admitted verdict needs a conversation key, or it can never queue")
+ }
+ return StateAdmitted, nil
+ case admission.StateBlocked, admission.StateDiscarded:
+ if v.Snapshot != nil {
+ return refuse("only an admitted verdict carries content")
+ }
+ return RecordState(v.State), nil
+ default:
+ return refuse(fmt.Sprintf("%q is not a state admission decides", v.State))
+ }
+}
+
+// explainVerdictRefusal says why a verdict changed nothing, reading the row in
+// the verdict's own transaction so the answer is the state that refused it.
+func explainVerdictRefusal(ctx context.Context, tx dbtx, v admission.Verdict) error {
+ var (
+ state string
+ revision int64
+ )
+ switch err := tx.QueryRowContext(ctx, `SELECT state, revision FROM events WHERE id = ?`, v.EventID).Scan(&state, &revision); {
+ case errors.Is(err, sql.ErrNoRows):
+ return fmt.Errorf("connector: verdict on %d: %w", v.EventID, ErrNoSuchRecord)
+ case err != nil:
+ return fmt.Errorf("connector: verdict on %d: %w", v.EventID, err)
+ }
+ // Moved on since the decision loaded it, by another decision or by
+ // dispatch: the newer state stands.
+ return fmt.Errorf("connector: verdict on %d loaded at revision %d, record is %s at revision %d: %w",
+ v.EventID, v.Revision, state, revision, admission.ErrAlreadyDecided)
+}
+
+// AdmissionOptions wires admission onto intake's queue and ledger.
+type AdmissionOptions struct {
+ Ledger *Ledger
+ Queue *Queue
+ Admitter *admission.Admitter
+ // Workers is the fetcher pool size; admission's default when zero.
+ Workers int
+ // Lines is the writer intake's pointer lines go through. Verdict lines
+ // share it, so the two never tear each other on one stdout.
+ Lines *ndjson.Writer
+ Logger *slog.Logger
+}
+
+// RunAdmission takes the ids intake hands over and decides each against the
+// ledger, until ctx ends. It returns what admission.Run returns.
+func RunAdmission(ctx context.Context, opts AdmissionOptions) error {
+ if opts.Ledger == nil || opts.Queue == nil {
+ return errors.New("connector: admission needs the ledger and the queue intake writes to")
+ }
+ records := opts.Ledger.Admission()
+ return admission.Run(ctx, admission.RunOptions{
+ Source: opts.Queue,
+ Records: records,
+ Admitter: opts.Admitter,
+ Committer: admission.NewCommitter(records),
+ Workers: opts.Workers,
+ LineWriter: opts.Lines,
+ Logger: opts.Logger,
+ })
+}
diff --git a/internal/connector/ledger_admission_test.go b/internal/connector/ledger_admission_test.go
new file mode 100644
index 000000000..fb6ca95b9
--- /dev/null
+++ b/internal/connector/ledger_admission_test.go
@@ -0,0 +1,742 @@
+package connector
+
+import (
+ "bytes"
+ "context"
+ "encoding/base64"
+ "encoding/json"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/basecamp/basecamp-sdk/go/pkg/basecamp"
+
+ "github.com/basecamp/basecamp-cli/internal/connector/admission"
+ "github.com/basecamp/basecamp-cli/internal/connector/ndjson"
+)
+
+// The join between intake's ledger and admission. Admission's own tests hold
+// its contract against a fake; these hold the real ledger to the same one.
+
+const (
+ adapterAgentID int64 = 52007412
+ adapterOperatorID int64 = 26909558
+ adapterBucketID int64 = 48699913
+)
+
+func seenRecord(t *testing.T, ledger *Ledger, id int64) Record {
+ t.Helper()
+ ctx := context.Background()
+ _, err := ledger.RecordSeen(ctx, testEvent(id), LanePoll)
+ require.NoError(t, err)
+ record, ok, err := ledger.Get(ctx, id)
+ require.NoError(t, err)
+ require.True(t, ok)
+ return record
+}
+
+func getRecord(t *testing.T, ledger *Ledger, id int64) Record {
+ t.Helper()
+ record, ok, err := ledger.Get(context.Background(), id)
+ require.NoError(t, err)
+ require.True(t, ok)
+ return record
+}
+
+func admittedVerdict(id, revision int64, key string) admission.Verdict {
+ return admission.Verdict{
+ EventID: id,
+ EventType: "comment.created",
+ BucketID: adapterBucketID,
+ RecordingID: 10304028972,
+ Revision: revision,
+ RequesterID: adapterOperatorID,
+ State: admission.StateAdmitted,
+ Trigger: admission.TriggerMentioned,
+ Acknowledge: true,
+ ConversationKey: key,
+ Reply: &admission.ReplyDestination{Kind: admission.ReplyComment, RecordingID: 10304028989},
+ Routed: true,
+ Route: "/work/connector",
+ Class: "internal",
+ RecordingURL: "https://app.basecamp.com/2914079/buckets/48699913/recordings/10304028972",
+ Snapshot: &admission.Snapshot{
+ Type: "Comment",
+ Title: "A comment",
+ AppURL: "https://app.basecamp.com/2914079/buckets/48699913/recordings/10304028972",
+ Content: "please look
",
+ UpdatedAt: time.Date(2026, 9, 16, 10, 0, 0, 0, time.UTC),
+ },
+ }
+}
+
+func blockedVerdict(id, revision int64, reason admission.Reason) admission.Verdict {
+ return admission.Verdict{
+ EventID: id, EventType: "comment.created", BucketID: adapterBucketID,
+ RecordingID: 10304028972, Revision: revision, RequesterID: adapterOperatorID,
+ State: admission.StateBlocked, Reason: reason,
+ }
+}
+
+func TestAdmissionLoadsAnUndecidedRecordAtItsRevision(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seen := seenRecord(t, ledger, 1)
+ store := ledger.Admission()
+
+ ev, ok, err := store.LoadUndecided(ctx, 1)
+ require.NoError(t, err)
+ require.True(t, ok)
+ assert.Equal(t, admission.Event{
+ ID: 1, EventType: "comment.created", BucketID: adapterBucketID, RecordingID: 10304028972,
+ CreatorID: adapterOperatorID, Revision: 0, SeenAt: seen.SeenAt,
+ }, ev)
+ assert.False(t, ev.SeenAt.IsZero(), "admission dates membership refusals from SeenAt")
+
+ // A blocked record is decided again, at the revision its verdict left.
+ _, err = store.Commit(ctx, blockedVerdict(1, 0, admission.ReasonReadFailed))
+ require.NoError(t, err)
+ ev, ok, err = store.LoadUndecided(ctx, 1)
+ require.NoError(t, err)
+ require.True(t, ok)
+ assert.Equal(t, int64(1), ev.Revision)
+
+ _, ok, err = store.LoadUndecided(ctx, 404)
+ require.NoError(t, err)
+ assert.False(t, ok, "an unknown id is skipped")
+}
+
+func TestAdmissionSkipsARecordPastDeciding(t *testing.T) {
+ for _, state := range []RecordState{StateAdmitted, StateQueued, StateDispatched, StateCompleted, StateDiscarded} {
+ t.Run(string(state), func(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ switch state {
+ case StateDiscarded:
+ require.NoError(t, ledger.SetState(ctx, 1, StateDiscarded, "untrusted_author"))
+ case StateCompleted:
+ require.NoError(t, reachTerminal(t, ledger, 7, StateCompleted))
+ require.NoError(t, ledger.SetState(ctx, 1, StateAdmitted, ""))
+ require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, ""))
+ require.NoError(t, ledger.SetState(ctx, 1, StateCompleted, ""))
+ case StateDispatched:
+ require.NoError(t, ledger.SetState(ctx, 1, StateAdmitted, ""))
+ require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, ""))
+ default:
+ require.NoError(t, ledger.SetState(ctx, 1, state, ""))
+ }
+
+ _, ok, err := ledger.Admission().LoadUndecided(ctx, 1)
+ require.NoError(t, err)
+ assert.False(t, ok)
+ })
+ }
+}
+
+func TestAdmissionCommitWritesTheVerdictOntoTheRecord(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ at := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC)
+ ledger.now = func() time.Time { return at }
+ seenRecord(t, ledger, 1)
+ v := admittedVerdict(1, 0, "recording:10304028989")
+
+ written, err := ledger.Admission().Commit(ctx, v)
+ require.NoError(t, err)
+ assert.Equal(t, admission.StateAdmitted, written)
+
+ record := getRecord(t, ledger, 1)
+ assert.Equal(t, StateAdmitted, record.State)
+ assert.Empty(t, record.Reason)
+ assert.Equal(t, int64(1), record.Revision)
+ d := record.Decision
+ require.NotNil(t, d.DecidedAt)
+ assert.True(t, at.Equal(*d.DecidedAt))
+ assert.Nil(t, d.BlockedAt)
+ assert.Nil(t, d.RetryAt)
+ assert.Equal(t, "mentioned", d.Trigger)
+ assert.True(t, d.Acknowledge)
+ assert.Equal(t, "recording:10304028989", d.ConversationKey)
+ assert.Equal(t, "comment", d.ReplyKind)
+ assert.Equal(t, int64(10304028989), d.ReplyRecordingID)
+ assert.True(t, d.Routed)
+ assert.Equal(t, "/work/connector", d.Route)
+ assert.Equal(t, "internal", d.Class)
+ assert.Equal(t, v.RecordingURL, d.RecordingURL)
+ assert.Equal(t, adapterOperatorID, d.RequesterID)
+
+ var snapshot admission.Snapshot
+ require.NoError(t, json.Unmarshal(d.Snapshot, &snapshot))
+ assert.Equal(t, *v.Snapshot, snapshot)
+}
+
+// Invariant 5 of admission: one verdict per event. The same decision arriving
+// twice — two fetchers, a restart — is written once.
+func TestAdmissionCommitIsOneVerdictPerEvent(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ store := ledger.Admission()
+
+ _, err := store.Commit(ctx, admittedVerdict(1, 0, "recording:9"))
+ require.NoError(t, err)
+ _, err = store.Commit(ctx, admittedVerdict(1, 0, "recording:9"))
+
+ require.ErrorIs(t, err, admission.ErrAlreadyDecided)
+ record := getRecord(t, ledger, 1)
+ assert.Equal(t, StateAdmitted, record.State, "not re-queued behind itself")
+ assert.Equal(t, int64(1), record.Revision)
+}
+
+func TestAdmissionAnOlderDecisionNeverOverwritesANewer(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ store := ledger.Admission()
+
+ // Two decisions loaded the record at revision 0; the blocked one lands
+ // first.
+ _, err := store.Commit(ctx, blockedVerdict(1, 0, admission.ReasonReadFailed))
+ require.NoError(t, err)
+
+ _, err = store.Commit(ctx, admittedVerdict(1, 0, "recording:9"))
+ require.ErrorIs(t, err, admission.ErrAlreadyDecided)
+ record := getRecord(t, ledger, 1)
+ assert.Equal(t, StateBlocked, record.State, "a blocked verdict stands against an older decision too")
+ assert.Equal(t, "read_failed", record.Reason)
+ assert.Empty(t, record.Decision.Snapshot)
+
+ // A decision loaded after it applies.
+ written, err := store.Commit(ctx, admittedVerdict(1, 1, "recording:9"))
+ require.NoError(t, err)
+ assert.Equal(t, admission.StateAdmitted, written)
+ assert.Empty(t, getRecord(t, ledger, 1).Reason)
+}
+
+// Every state change bumps the revision, not only admission's: a decision
+// loaded before anything else moved the record is stale.
+func TestAdmissionAnyMoveStalesALoadedDecision(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ store := ledger.Admission()
+ ev, ok, err := store.LoadUndecided(ctx, 1)
+ require.NoError(t, err)
+ require.True(t, ok)
+
+ require.NoError(t, ledger.SetState(ctx, 1, StateBlocked, "no_route"))
+
+ _, err = store.Commit(ctx, admittedVerdict(1, ev.Revision, "recording:9"))
+ require.ErrorIs(t, err, admission.ErrAlreadyDecided)
+ assert.Equal(t, StateBlocked, getRecord(t, ledger, 1).State)
+}
+
+// A record admission already decided, or dispatch has moved on, is never
+// decided again — even by a verdict carrying its current revision.
+func TestAdmissionNeverDecidesARecordPastDeciding(t *testing.T) {
+ cases := map[string]func(*testing.T, *Ledger){
+ "admitted": func(t *testing.T, l *Ledger) {
+ require.NoError(t, l.SetState(context.Background(), 1, StateAdmitted, ""))
+ },
+ "dispatched": func(t *testing.T, l *Ledger) {
+ require.NoError(t, l.SetState(context.Background(), 1, StateAdmitted, ""))
+ require.NoError(t, l.SetState(context.Background(), 1, StateDispatched, ""))
+ },
+ "discarded": func(t *testing.T, l *Ledger) {
+ require.NoError(t, l.SetState(context.Background(), 1, StateDiscarded, "by_operator"))
+ },
+ }
+ for name, reach := range cases {
+ t.Run(name, func(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ reach(t, ledger)
+ before := getRecord(t, ledger, 1)
+
+ for _, v := range []admission.Verdict{
+ admittedVerdict(1, before.Revision, "recording:9"),
+ blockedVerdict(1, before.Revision, admission.ReasonReadFailed),
+ } {
+ _, err := ledger.Admission().Commit(ctx, v)
+ require.ErrorIs(t, err, admission.ErrAlreadyDecided)
+ }
+ after := getRecord(t, ledger, 1)
+ assert.Equal(t, before.State, after.State)
+ assert.Equal(t, before.Revision, after.Revision)
+ })
+ }
+}
+
+func TestAdmissionCommitOnAMissingRecord(t *testing.T) {
+ _, err := newTestLedger(t).Admission().Commit(context.Background(), blockedVerdict(404, 0, admission.ReasonReadFailed))
+ assert.ErrorIs(t, err, ErrNoSuchRecord)
+ assert.NotErrorIs(t, err, admission.ErrAlreadyDecided)
+}
+
+// Admitted or queued is the ledger's decision, taken against the conversation
+// as it stands in the verdict's own transaction.
+func TestAdmissionQueuesBehindALiveConversation(t *testing.T) {
+ const key = "recording:10304028989"
+ cases := []struct {
+ name string
+ other func(*testing.T, *Ledger)
+ want admission.State
+ }{
+ {"nothing else on the key", nil, admission.StateAdmitted},
+ {"an admitted record becomes the task", func(t *testing.T, l *Ledger) {
+ _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key))
+ require.NoError(t, err)
+ }, admission.StateQueued},
+ {"a dispatched record is a running task", func(t *testing.T, l *Ledger) {
+ _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key))
+ require.NoError(t, err)
+ require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, ""))
+ }, admission.StateQueued},
+ {"a queued record alone is not a task", func(t *testing.T, l *Ledger) {
+ _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key))
+ require.NoError(t, err)
+ _, err = l.Admission().Commit(context.Background(), admittedVerdict(3, 0, key))
+ require.NoError(t, err)
+ require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, ""))
+ require.NoError(t, l.SetState(context.Background(), 2, StateCompleted, ""))
+ require.Equal(t, StateQueued, getRecord(t, l, 3).State)
+ }, admission.StateAdmitted},
+ {"a completed task is not live", func(t *testing.T, l *Ledger) {
+ _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, key))
+ require.NoError(t, err)
+ require.NoError(t, l.SetState(context.Background(), 2, StateDispatched, ""))
+ require.NoError(t, l.SetState(context.Background(), 2, StateCompleted, ""))
+ }, admission.StateAdmitted},
+ {"a blocked record is not a task", func(t *testing.T, l *Ledger) {
+ v := admittedVerdict(2, 0, key)
+ v.State, v.Reason, v.Snapshot = admission.StateBlocked, admission.ReasonNoRoute, nil
+ _, err := l.Admission().Commit(context.Background(), v)
+ require.NoError(t, err)
+ }, admission.StateAdmitted},
+ {"a live task on another conversation", func(t *testing.T, l *Ledger) {
+ _, err := l.Admission().Commit(context.Background(), admittedVerdict(2, 0, "recording:1"))
+ require.NoError(t, err)
+ }, admission.StateAdmitted},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ ledger := newTestLedger(t)
+ seenRecord(t, ledger, 1)
+ seenRecord(t, ledger, 2)
+ seenRecord(t, ledger, 3)
+ if tc.other != nil {
+ tc.other(t, ledger)
+ }
+
+ written, err := ledger.Admission().Commit(context.Background(), admittedVerdict(1, 0, key))
+ require.NoError(t, err)
+ assert.Equal(t, tc.want, written)
+ assert.Equal(t, RecordState(tc.want), getRecord(t, ledger, 1).State)
+ if tc.name == "a queued record alone is not a task" {
+ // Liveness decides only the incoming record. The queued one
+ // is not moved, expired or re-decided by it.
+ third := getRecord(t, ledger, 3)
+ assert.Equal(t, StateQueued, third.State)
+ assert.Equal(t, int64(1), third.Revision)
+ }
+ })
+ }
+}
+
+// Only an admitted verdict carries content, and a verdict the ledger refuses
+// writes nothing at all.
+func TestAdmissionRefusesAMalformedVerdictWithoutWriting(t *testing.T) {
+ blockedWithContent := blockedVerdict(1, 0, admission.ReasonNoRoute)
+ blockedWithContent.Snapshot = admittedVerdict(1, 0, "k").Snapshot
+ discardedWithContent := blockedWithContent
+ discardedWithContent.State, discardedWithContent.Reason = admission.StateDiscarded, admission.ReasonStale
+ admittedWithout := admittedVerdict(1, 0, "recording:9")
+ admittedWithout.Snapshot = nil
+ admittedNoKey := admittedVerdict(1, 0, "")
+ queued := admittedVerdict(1, 0, "recording:9")
+ queued.State, queued.Snapshot = admission.StateQueued, nil
+ blockedNoReason := blockedVerdict(1, 0, "")
+
+ for name, v := range map[string]admission.Verdict{
+ "blocked with content": blockedWithContent,
+ "discarded with content": discardedWithContent,
+ "admitted without content": admittedWithout,
+ "admitted without a key": admittedNoKey,
+ "queued is not a verdict": queued,
+ "blocked without a reason": blockedNoReason,
+ } {
+ t.Run(name, func(t *testing.T) {
+ ledger := newTestLedger(t)
+ seenRecord(t, ledger, 1)
+
+ _, err := ledger.Admission().Commit(context.Background(), v)
+
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, admission.ErrAlreadyDecided)
+ record := getRecord(t, ledger, 1)
+ assert.Equal(t, StateSeen, record.State)
+ assert.Equal(t, int64(0), record.Revision)
+ assert.Nil(t, record.Decision.DecidedAt)
+ assert.Empty(t, record.Decision.Snapshot)
+ })
+ }
+}
+
+// A throttled verdict keeps the server's deadline, and the retry window counts
+// from the first of a run of blocked verdicts.
+func TestAdmissionKeepsTheBlockedScheduleInputs(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ store := ledger.Admission()
+ t0 := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC)
+ ledger.now = func() time.Time { return t0 }
+ seenRecord(t, ledger, 1)
+
+ throttled := blockedVerdict(1, 0, admission.ReasonThrottled)
+ throttled.RetryAt = t0.Add(7 * time.Minute)
+ _, err := store.Commit(ctx, throttled)
+ require.NoError(t, err)
+ d := getRecord(t, ledger, 1).Decision
+ require.NotNil(t, d.RetryAt)
+ assert.True(t, throttled.RetryAt.Equal(*d.RetryAt))
+ require.NotNil(t, d.BlockedAt)
+ assert.True(t, t0.Equal(*d.BlockedAt))
+
+ // Re-run ten minutes later and blocked again, on a read this time.
+ t1 := t0.Add(10 * time.Minute)
+ ledger.now = func() time.Time { return t1 }
+ _, err = store.Commit(ctx, blockedVerdict(1, 1, admission.ReasonReadFailed))
+ require.NoError(t, err)
+ d = getRecord(t, ledger, 1).Decision
+ assert.Nil(t, d.RetryAt, "only a throttled verdict has a deadline")
+ require.NotNil(t, d.BlockedAt)
+ assert.True(t, t0.Equal(*d.BlockedAt), "the window still counts from the first block")
+ require.NotNil(t, d.DecidedAt)
+ assert.True(t, t1.Equal(*d.DecidedAt), "the last attempt is the latest verdict")
+
+ // NextBlockedRetry reads what the ledger kept.
+ next, ok := admission.NextBlockedRetry(admission.Reason(getRecord(t, ledger, 1).Reason), *d.BlockedAt, *d.DecidedAt, time.Time{})
+ require.True(t, ok)
+ assert.True(t, t1.Add(admission.BlockedRetryInterval).Equal(next))
+
+ // Admitted at last: no longer blocked, and a deadline is a blocked
+ // record's alone.
+ admitted := admittedVerdict(1, 2, "recording:9")
+ admitted.RetryAt = t1.Add(time.Hour)
+ _, err = store.Commit(ctx, admitted)
+ require.NoError(t, err)
+ d = getRecord(t, ledger, 1).Decision
+ assert.Nil(t, d.BlockedAt)
+ assert.Nil(t, d.RetryAt)
+}
+
+// The blocked schedule's inputs belong to every state write, not only to a
+// verdict: a record moved out of blocked by anything forgets when it was
+// blocked, one moved back in starts a new window, and a record parked on
+// blocked or discarded carries no content.
+func TestEveryMoveKeepsTheBlockedScheduleInputs(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ store := ledger.Admission()
+ t0 := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC)
+ ledger.now = func() time.Time { return t0 }
+ seenRecord(t, ledger, 1)
+ throttled := blockedVerdict(1, 0, admission.ReasonThrottled)
+ throttled.RetryAt = t0.Add(time.Hour)
+ _, err := store.Commit(ctx, throttled)
+ require.NoError(t, err)
+
+ // Out of blocked by a plain state write: no window, no deadline.
+ t1 := t0.Add(72 * time.Hour)
+ ledger.now = func() time.Time { return t1 }
+ require.NoError(t, ledger.SetState(ctx, 1, StateAdmitted, ""))
+ d := getRecord(t, ledger, 1).Decision
+ assert.Nil(t, d.BlockedAt, "a record that left blocked is not blocked")
+ assert.Nil(t, d.RetryAt)
+
+ // Back into blocked after a dispatch: a new window from now, and the
+ // verdict that follows keeps it.
+ require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, ""))
+ require.NoError(t, ledger.SetState(ctx, 1, StateBlocked, "read_failed"))
+ record := getRecord(t, ledger, 1)
+ require.NotNil(t, record.Decision.BlockedAt)
+ assert.True(t, t1.Equal(*record.Decision.BlockedAt))
+ t2 := t1.Add(10 * time.Minute)
+ ledger.now = func() time.Time { return t2 }
+ _, err = store.Commit(ctx, blockedVerdict(1, record.Revision, admission.ReasonReadFailed))
+ require.NoError(t, err)
+ d = getRecord(t, ledger, 1).Decision
+ require.NotNil(t, d.BlockedAt)
+ _, ok := admission.NextBlockedRetry(admission.ReasonReadFailed, *d.BlockedAt, *d.DecidedAt, time.Time{})
+ assert.True(t, ok, "a record blocked again after a redispatch still gets its timed retries")
+
+ // A deadline is a blocked record's alone.
+ _, err = ledger.move(ctx, ledger.db, transition{id: 1, state: StateBlocked, reason: "read_failed", retryAt: t2})
+ require.NoError(t, err)
+ _, err = ledger.move(ctx, ledger.db, transition{id: 1, state: StateAdmitted, retryAt: t2})
+ require.Error(t, err)
+}
+
+func TestAMoveToBlockedOrDiscardedDropsTheSnapshot(t *testing.T) {
+ for _, target := range []RecordState{StateBlocked, StateDiscarded} {
+ t.Run(string(target), func(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ _, err := ledger.Admission().Commit(ctx, admittedVerdict(1, 0, "recording:9"))
+ require.NoError(t, err)
+ require.NotEmpty(t, getRecord(t, ledger, 1).Decision.Snapshot)
+
+ require.NoError(t, ledger.SetState(ctx, 1, target, "by_operator"))
+
+ assert.Empty(t, getRecord(t, ledger, 1).Decision.Snapshot)
+ })
+ }
+
+ t.Run("dispatch keeps it", func(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ seenRecord(t, ledger, 1)
+ _, err := ledger.Admission().Commit(ctx, admittedVerdict(1, 0, "recording:9"))
+ require.NoError(t, err)
+ require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, ""))
+ require.NoError(t, ledger.SetState(ctx, 1, StateCompleted, ""))
+ assert.NotEmpty(t, getRecord(t, ledger, 1).Decision.Snapshot)
+ })
+}
+
+// The conversation read uses the conversation index rather than scanning
+// every record in a live state.
+func TestTheConversationReadUsesItsIndex(t *testing.T) {
+ ledger := newTestLedger(t)
+ var plan strings.Builder
+ rows, err := ledger.db.QueryContext(context.Background(), "EXPLAIN QUERY PLAN "+liveConversation, "recording:9", "admitted", "dispatched")
+ require.NoError(t, err)
+ defer rows.Close()
+ for rows.Next() {
+ var id, parent, unused int
+ var detail string
+ require.NoError(t, rows.Scan(&id, &parent, &unused, &detail))
+ plan.WriteString(detail + "\n")
+ }
+ require.NoError(t, rows.Err())
+ assert.Contains(t, plan.String(), "events_conversation")
+}
+
+// Two fetchers deciding one event at once: exactly one verdict is written.
+func TestAdmissionConcurrentDecisionsWriteOne(t *testing.T) {
+ ledger := newTestLedger(t)
+ seenRecord(t, ledger, 1)
+ store := ledger.Admission()
+
+ const racers = 8
+ results := make([]error, racers)
+ var wg sync.WaitGroup
+ for i := range racers {
+ wg.Go(func() {
+ v := admittedVerdict(1, 0, "recording:9")
+ if i%2 == 1 {
+ v = blockedVerdict(1, 0, admission.ReasonReadFailed)
+ }
+ _, results[i] = store.Commit(context.Background(), v)
+ })
+ }
+ wg.Wait()
+
+ applied := 0
+ for _, err := range results {
+ if err == nil {
+ applied++
+ continue
+ }
+ require.ErrorIs(t, err, admission.ErrAlreadyDecided)
+ }
+ assert.Equal(t, 1, applied)
+ assert.Equal(t, int64(1), getRecord(t, ledger, 1).Revision)
+}
+
+// Retention takes the verdict's content with the pointer's.
+func TestDropContentTakesTheVerdictToo(t *testing.T) {
+ ledger := newTestLedger(t)
+ ctx := context.Background()
+ at := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC)
+ ledger.now = func() time.Time { return at }
+ seenRecord(t, ledger, 1)
+ _, err := ledger.Admission().Commit(ctx, admittedVerdict(1, 0, "recording:9"))
+ require.NoError(t, err)
+ require.NoError(t, ledger.SetState(ctx, 1, StateDispatched, ""))
+ require.NoError(t, ledger.SetState(ctx, 1, StateCompleted, ""))
+
+ dropped, err := ledger.DropContent(ctx, at.Add(time.Hour), at.Add(time.Hour))
+ require.NoError(t, err)
+ require.Equal(t, 1, dropped)
+
+ record := getRecord(t, ledger, 1)
+ assert.Equal(t, StateCompleted, record.State)
+ assert.Equal(t, Decision{DecidedAt: record.Decision.DecidedAt}, record.Decision,
+ "only the tombstone's timestamps survive")
+}
+
+// Migration 4 carries a ledger written before it: its records load at
+// revision 0 and are decided like any other.
+func TestMigrationFourCarriesAnEarlierLedger(t *testing.T) {
+ path := t.TempDir() + "/state/connector.db"
+ all := migrations
+ migrations = all[:3]
+ old, err := OpenLedger(path)
+ migrations = all
+ require.NoError(t, err)
+ _, err = old.RecordSeen(context.Background(), testEvent(1), LanePoll)
+ require.NoError(t, err)
+ require.NoError(t, old.Close())
+
+ ledger, err := OpenLedger(path)
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = ledger.Close() })
+ version, err := ledger.SchemaVersion(context.Background())
+ require.NoError(t, err)
+ assert.Equal(t, 4, version)
+
+ ev, ok, err := ledger.Admission().LoadUndecided(context.Background(), 1)
+ require.NoError(t, err)
+ require.True(t, ok)
+ assert.Equal(t, int64(0), ev.Revision)
+ _, err = ledger.Admission().Commit(context.Background(), admittedVerdict(1, ev.Revision, "recording:9"))
+ require.NoError(t, err)
+ assert.Equal(t, StateAdmitted, getRecord(t, ledger, 1).State)
+}
+
+// The handoff end to end: intake records and offers, admission takes from the
+// same queue, decides against the real ledger, and reports on the shared
+// writer without content.
+func TestRunAdmissionDecidesWhatIntakeHandsOver(t *testing.T) {
+ ledger := newTestLedger(t)
+ queue, err := NewQueue(10, 100)
+ require.NoError(t, err)
+ var out bytes.Buffer
+ lines := ndjson.NewWriter(&out)
+
+ const secret = "the instruction itself"
+ reads := &adapterReads{summaries: map[int64]*basecamp.RecordingSummary{}}
+ for _, id := range []int64{501, 502} {
+ reads.summaries[id] = &basecamp.RecordingSummary{
+ ID: id, Status: "active", Type: "Todo", Title: "A to-do",
+ AppURL: "https://app.basecamp.com/2914079/buckets/48699913/todos/" + strconv.FormatInt(id, 10),
+ Bucket: &basecamp.Bucket{ID: adapterBucketID},
+ Creator: &basecamp.Person{ID: adapterOperatorID},
+ Content: mentionMarkup(adapterAgentID) + secret,
+ MentionedPersonIDs: []int64{adapterAgentID},
+ UpdatedAt: time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC),
+ }
+ }
+ admitter, err := admission.NewAdmitter(admission.Policy{
+ AgentID: adapterAgentID,
+ Trust: admission.Trust{Mode: admission.TrustOperator, OperatorID: adapterOperatorID},
+ Projects: map[int64]admission.Route{adapterBucketID: {Path: "/work/connector", Class: "internal"}},
+ }, admission.Reads{Summaries: reads, Subscriptions: reads, Assignments: reads})
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() {
+ done <- RunAdmission(ctx, AdmissionOptions{Ledger: ledger, Queue: queue, Admitter: admitter, Workers: 2, Lines: lines})
+ }()
+
+ // Two to-dos, then a third event on the first one's conversation: a
+ // to-do's conversation is itself, so the third queues behind the first.
+ for i, recording := range []int64{501, 502, 501} {
+ ev := testEvent(int64(i + 1))
+ ev.EventType, ev.Kind, ev.RecordingID = "todo.created", "todo_created", recording
+ _, err := ledger.RecordSeen(ctx, ev, LanePoll)
+ require.NoError(t, err)
+ require.NoError(t, queue.Offer(ctx, ev.ID))
+ require.Eventually(t, func() bool {
+ return getRecord(t, ledger, ev.ID).State != StateSeen
+ }, 5*time.Second, 10*time.Millisecond)
+ }
+ cancel()
+ require.NoError(t, <-done)
+
+ assert.Equal(t, StateAdmitted, getRecord(t, ledger, 1).State)
+ assert.Equal(t, StateAdmitted, getRecord(t, ledger, 2).State)
+ third := getRecord(t, ledger, 3)
+ assert.Equal(t, StateQueued, third.State)
+ assert.Contains(t, string(third.Decision.Snapshot), secret, "the queued follow-up keeps its instruction")
+
+ assert.NotContains(t, out.String(), secret, "no content on the wire")
+ // Two workers: a verdict is visible in the ledger before its line is
+ // written, so lines arrive in no promised order. What is promised is one
+ // whole line per verdict, reporting the state the ledger wrote.
+ states := map[int64]admission.State{}
+ for _, raw := range strings.Split(strings.TrimSpace(out.String()), "\n") {
+ var line admission.Line
+ require.NoError(t, json.Unmarshal([]byte(raw), &line), "every line is whole JSON: %q", raw)
+ _, dup := states[line.EventID]
+ require.False(t, dup, "one line per verdict, event %d twice", line.EventID)
+ states[line.EventID] = line.State
+ }
+ assert.Equal(t, map[int64]admission.State{
+ 1: admission.StateAdmitted,
+ 2: admission.StateAdmitted,
+ 3: admission.StateQueued,
+ }, states, "the line reports what the ledger wrote")
+}
+
+func TestRunAdmissionNeedsTheLedgerAndQueue(t *testing.T) {
+ admitter, err := admission.NewAdmitter(admission.Policy{
+ AgentID: adapterAgentID,
+ Trust: admission.Trust{Mode: admission.TrustOperator, OperatorID: adapterOperatorID},
+ }, admission.Reads{Summaries: &adapterReads{}, Subscriptions: &adapterReads{}, Assignments: &adapterReads{}})
+ require.NoError(t, err)
+ queue, err := NewQueue(1, 1)
+ require.NoError(t, err)
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+
+ for name, opts := range map[string]AdmissionOptions{
+ "no ledger": {Queue: queue, Admitter: admitter},
+ "no queue": {Ledger: newTestLedger(t), Admitter: admitter},
+ } {
+ t.Run(name, func(t *testing.T) {
+ err := RunAdmission(ctx, opts)
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, context.DeadlineExceeded, "refused at once, not run until the deadline")
+ })
+ }
+}
+
+// adapterReads answers admission's reads for the handoff test.
+type adapterReads struct {
+ summaries map[int64]*basecamp.RecordingSummary
+}
+
+func (r *adapterReads) Summarize(_ context.Context, ref basecamp.RecordingRef) (*basecamp.RecordingSummary, error) {
+ s, ok := r.summaries[ref.RecordingID]
+ if !ok {
+ return nil, &basecamp.Error{Code: basecamp.CodeNotFound, Message: "not found"}
+ }
+ copied := *s
+ return &copied, nil
+}
+
+func (r *adapterReads) Subscribed(context.Context, int64) (bool, error) { return false, nil }
+
+func (r *adapterReads) AddedPersonIDs(context.Context, int64, int64) ([]int64, bool, error) {
+ return nil, false, nil
+}
+
+// mentionMarkup is a mention attachment naming id only inside its sgid, the
+// way Basecamp renders one.
+func mentionMarkup(id int64) string {
+ payload := `{"_rails":{"data":"gid://bc3/Person/` + strconv.FormatInt(id, 10) + `","pur":"attachable"}}`
+ sgid := base64.RawURLEncoding.EncodeToString([]byte(payload))
+ return ``
+}
diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go
index 1867f7213..54d6f3281 100644
--- a/internal/connector/ledger_events.go
+++ b/internal/connector/ledger_events.go
@@ -33,6 +33,43 @@ type Record struct {
SeenAt time.Time
UpdatedAt time.Time
ContentDropped bool
+
+ // Revision counts every state write applied to the record since intake
+ // recorded it, a repeat of the state it already has included. A decision
+ // applies only at the revision it loaded.
+ Revision int64
+ // Decision is admission's latest verdict on the record; its zero value
+ // until one is written.
+ Decision Decision
+}
+
+// Decision is what admission wrote onto a record with its verdict. The pointer
+// fields stay intake's; these are admission's, and dispatch starts a task from
+// them.
+type Decision struct {
+ // DecidedAt is when the latest verdict was written.
+ DecidedAt *time.Time
+ // BlockedAt is when the record entered its current run of blocked
+ // verdicts; nil unless it is blocked.
+ BlockedAt *time.Time
+ // RetryAt is a throttled verdict's server deadline; nil otherwise.
+ RetryAt *time.Time
+
+ Trigger string
+ Acknowledge bool
+ ConversationKey string
+ ReplyKind string
+ ReplyRecordingID int64
+ Routed bool
+ Route string
+ Class string
+ RecordingURL string
+ RequesterID int64
+ // Snapshot is the recording's content as admission read it, JSON. An
+ // admitted verdict writes it, as admitted or queued; it stays through
+ // dispatch and completion until retention drops it, and any move to
+ // blocked or discarded clears it.
+ Snapshot json.RawMessage
}
// EffectivePerformer is the id the performers/exclude_performers filters
@@ -168,8 +205,14 @@ func (l *Ledger) CountInState(ctx context.Context, state RecordState) (int, erro
// edges back into the working states. Completed and discarded have none.
var lifecycle = map[RecordState][]RecordState{
// seen to queued is one edge, not two: admission commits an admitted
- // verdict AS queued when the conversation is already live, so the record
- // never passes through admitted at all.
+ // verdict AS queued when the conversation is already live — another
+ // record on it admitted or dispatched, admission.Ledger's definition — so
+ // the record never passes through admitted at all. Deciding that moves
+ // only the incoming record; a queued record leaves queued by dispatch.
+ //
+ // A blocked record's edges back into the working states are for the
+ // lifecycle's own bookkeeping. It returns to work with content only
+ // through a new verdict, because a move to blocked drops the snapshot.
StateSeen: {StateAdmitted, StateQueued, StateBlocked, StateDiscarded},
StateAdmitted: {StateQueued, StateDispatched, StateBlocked, StateDiscarded},
StateQueued: {StateDispatched, StateBlocked, StateDiscarded},
@@ -211,47 +254,135 @@ var ErrNotATransition = errors.New("not a transition the ledger's lifecycle allo
// two statements is a race, and this is the guarantee that a finished record
// stays finished.
func (l *Ledger) SetState(ctx context.Context, id int64, state RecordState, reason string) error {
- switch state {
+ moved, err := l.move(ctx, l.db, transition{id: id, state: state, reason: reason})
+ if err != nil {
+ return err
+ }
+ if !moved {
+ return l.explainRefusal(ctx, id, state)
+ }
+ return nil
+}
+
+// dbtx is what a transition runs against: the ledger's handle, or a
+// transaction a caller already holds so the move commits with its other
+// writes.
+type dbtx interface {
+ ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+ QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
+}
+
+// transition is one state change and what is written with it.
+type transition struct {
+ id int64
+ state RecordState
+ reason string
+ // from narrows the states the record may be leaving. Empty means every
+ // state the lifecycle lets reach state; a narrowing can only remove edges,
+ // never add one.
+ from []RecordState
+ // revision, when set, applies the move only while the record is still at
+ // that revision.
+ revision *int64
+ // retryAt is a blocked record's not-before deadline, stored as retry_at;
+ // the zero time stores none. Only a move to blocked may carry one.
+ retryAt time.Time
+ // set is further columns written in the same statement.
+ set []assignment
+}
+
+// assignment is one further column a transition writes. column is this
+// package's own constant, never input; value is bound.
+type assignment struct {
+ column string
+ value any
+}
+
+// move is the ledger's one state-changing write. Every change of state goes
+// through it, so every change meets the lifecycle, bumps the revision, and
+// keeps the retention clock where a repeat must leave it. It reports whether
+// the record moved; a refusal is the caller's to explain.
+//
+// revision is bumped on every applied write, a repeat included. It is what a
+// decision loaded earlier is compared against, and any write since that load
+// is a reason the decision may no longer hold.
+func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) {
+ switch t.state {
case StateBlocked, StateDiscarded:
- if reason == "" {
- return fmt.Errorf("connector: set state of %d: a %s record needs a reason", id, state)
+ if t.reason == "" {
+ return false, fmt.Errorf("connector: set state of %d: a %s record needs a reason", t.id, t.state)
}
case StateSeen, StateAdmitted, StateQueued, StateDispatched, StateCompleted:
- if reason != "" {
- return fmt.Errorf("connector: set state of %d: a %s record takes no reason", id, state)
+ if t.reason != "" {
+ return false, fmt.Errorf("connector: set state of %d: a %s record takes no reason", t.id, t.state)
}
default:
// A state outside the lifecycle is a row no recovery scan looks for.
- return fmt.Errorf("connector: set state of %d: %q is not a ledger state", id, state)
+ return false, fmt.Errorf("connector: set state of %d: %q is not a ledger state", t.id, t.state)
}
- froms := enterableFrom(state)
- args := []any{string(state), reason, string(state), l.timestamp(), id}
- for _, from := range froms {
- args = append(args, from)
+ if !t.retryAt.IsZero() && t.state != StateBlocked {
+ return false, fmt.Errorf("connector: set state of %d: only a blocked record has a retry deadline", t.id)
}
- // The only thing concatenated is a list of "?" as long as the lifecycle's
- // own edge list. Every value is bound.
- //nolint:gosec // G202: placeholders, not values
- //
+ froms := enterableFrom(t.state)
+ if len(t.from) > 0 {
+ froms = slices.DeleteFunc(froms, func(from string) bool {
+ return !slices.Contains(t.from, RecordState(from))
+ })
+ }
+
// updated_at is left alone when the state does not change. It is the
// retention clock DropContent reads, and a repeated write of the state a
// record already has would silently restart the window on a finished
- // record.
- query := `UPDATE events SET state = ?, reason = ?,
- updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END
-WHERE id = ? AND state IN (` + strings.TrimSuffix(strings.Repeat("?, ", len(froms)), ", ") + `)`
- res, err := l.db.ExecContext(ctx, query, args...)
+ // record. Every right-hand side reads the row as it was before this
+ // statement, which is SQLite's rule for UPDATE.
+ //
+ // The blocked schedule's inputs are the move's too, so no path into or out
+ // of blocked can leave them stale. blocked_at is when the record entered
+ // its current run of blocked states — cleared on leaving, so it is set
+ // exactly on entering and kept across a repeat — because the retry
+ // window counts from it.
+ // retry_at holds only for the move that set it. And a record moving to
+ // blocked or discarded loses its snapshot: only a record on its way to a
+ // worker carries content.
+ now := l.timestamp()
+ var retryAt any
+ if !t.retryAt.IsZero() {
+ retryAt = stamp(t.retryAt)
+ }
+ var query strings.Builder
+ query.WriteString(`UPDATE events SET state = ?, reason = ?, revision = revision + 1,
+ updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END,
+ blocked_at = CASE WHEN ? <> 'blocked' THEN NULL ELSE COALESCE(blocked_at, ?) END,
+ retry_at = ?,
+ snapshot = CASE WHEN ? IN ('blocked', 'discarded') THEN NULL ELSE snapshot END`)
+ args := []any{string(t.state), t.reason, string(t.state), now, string(t.state), now, retryAt, string(t.state)}
+ for _, a := range t.set {
+ query.WriteString(", " + a.column + " = ?")
+ args = append(args, a.value)
+ }
+ query.WriteString(" WHERE id = ?")
+ args = append(args, t.id)
+ if t.revision != nil {
+ query.WriteString(" AND revision = ?")
+ args = append(args, *t.revision)
+ }
+ query.WriteString(" AND state IN (" + strings.TrimSuffix(strings.Repeat("?, ", len(froms)), ", ") + ")")
+ for _, from := range froms {
+ args = append(args, from)
+ }
+
+ // Concatenated are column names this package declares,
+ // and a list of "?" as long as the lifecycle's own edge list. Every value
+ // is bound.
+ res, err := db.ExecContext(ctx, query.String(), args...) //nolint:gosec // G202: constants and placeholders, not values
if err != nil {
- return fmt.Errorf("connector: set state of %d: %w", id, err)
+ return false, fmt.Errorf("connector: set state of %d: %w", t.id, err)
}
affected, err := res.RowsAffected()
if err != nil {
- return fmt.Errorf("connector: set state of %d: %w", id, err)
+ return false, fmt.Errorf("connector: set state of %d: %w", t.id, err)
}
- if affected == 0 {
- return l.explainRefusal(ctx, id, state)
- }
- return nil
+ return affected > 0, nil
}
// explainRefusal says why an update changed nothing: there is no such record,
@@ -286,7 +417,10 @@ func (l *Ledger) DropContent(ctx context.Context, discardedBefore, completedBefo
UPDATE events
SET details = NULL, event_type = '', kind = '', action = '', bucket_id = 0,
creator_id = 0, performed_by_id = NULL, recording_id = 0, actor_type = '',
- visible_to_clients = NULL, content_dropped = 1, updated_at = updated_at
+ visible_to_clients = NULL, content_dropped = 1, updated_at = updated_at,
+ snapshot = NULL, trigger_name = '', acknowledge = 0, conversation_key = '',
+ reply_kind = '', reply_recording_id = 0, routed = 0, route = '', class = '',
+ recording_url = '', requester_id = 0
WHERE content_dropped = 0
AND ((state = ? AND updated_at < ?) OR (state = ? AND updated_at < ?))`,
string(StateDiscarded), stamp(discardedBefore),
@@ -304,7 +438,10 @@ WHERE content_dropped = 0
const selectRecords = `
SELECT id, state, reason, lane, event_type, kind, action, bucket_id, creator_id,
performed_by_id, recording_id, details, actor_type, visible_to_clients,
- created_at, seen_at, updated_at, content_dropped
+ created_at, seen_at, updated_at, content_dropped, revision, decided_at,
+ blocked_at, retry_at, trigger_name, acknowledge, conversation_key,
+ reply_kind, reply_recording_id, routed, route, class, recording_url,
+ requester_id, snapshot
FROM events`
func scanRecords(rows *sql.Rows) ([]Record, error) {
@@ -320,11 +457,19 @@ func scanRecords(rows *sql.Rows) ([]Record, error) {
contentDropped int
performedBy sql.NullInt64
visibleToClients sql.NullBool
+ decidedAt, blockedAt sql.NullString
+ retryAt sql.NullString
+ acknowledge, routed int
+ snapshot []byte
+ d = &r.Decision
)
if err := rows.Scan(&r.ID, &state, &r.Reason, &lane, &r.EventType, &r.Kind,
&r.Action, &r.BucketID, &r.CreatorID, &performedBy, &r.RecordingID,
&details, &r.ActorType, &visibleToClients, &createdAt, &seenAt,
- &updatedAt, &contentDropped); err != nil {
+ &updatedAt, &contentDropped, &r.Revision, &decidedAt, &blockedAt,
+ &retryAt, &d.Trigger, &acknowledge, &d.ConversationKey, &d.ReplyKind,
+ &d.ReplyRecordingID, &routed, &d.Route, &d.Class, &d.RecordingURL,
+ &d.RequesterID, &snapshot); err != nil {
return nil, fmt.Errorf("connector: scan event record: %w", err)
}
r.State = RecordState(state)
@@ -351,6 +496,23 @@ func scanRecords(rows *sql.Rows) ([]Record, error) {
return nil, err
}
r.ContentDropped = contentDropped != 0
+ d.Acknowledge, d.Routed = acknowledge != 0, routed != 0
+ if len(snapshot) > 0 {
+ d.Snapshot = json.RawMessage(snapshot)
+ }
+ for _, stamped := range []struct {
+ raw sql.NullString
+ to **time.Time
+ }{{decidedAt, &d.DecidedAt}, {blockedAt, &d.BlockedAt}, {retryAt, &d.RetryAt}} {
+ if !stamped.raw.Valid {
+ continue
+ }
+ at, err := parseStamp(stamped.raw.String)
+ if err != nil {
+ return nil, err
+ }
+ *stamped.to = &at
+ }
records = append(records, r)
}
return records, rows.Err()