diff --git a/go/internal/store/db/forge_state_transitions.sql.go b/go/internal/store/db/forge_state_transitions.sql.go new file mode 100644 index 00000000..88dc68ea --- /dev/null +++ b/go/internal/store/db/forge_state_transitions.sql.go @@ -0,0 +1,102 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: forge_state_transitions.sql + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const consumeStateTransition = `-- name: ConsumeStateTransition :one +UPDATE forge_state_transitions + SET consumed_at = now() + WHERE forge_provider = $1 + AND forge_host = $2 + AND repo = $3 + AND kind = $4 + AND number = $5 + AND state = $6 + AND consumed_at IS NULL + AND written_at >= $7 +RETURNING agent_account_id +` + +type ConsumeStateTransitionParams struct { + ForgeProvider int16 + ForgeHost string + Repo string + Kind int16 + Number int64 + State string + WrittenAt pgtype.Timestamptz +} + +// Single-statement clear-and-return: the row is claimed and its actor returned +// in ONE UPDATE, so a memo attributes at most one event and a concurrent second +// reader matches nothing (consumed_at is no longer NULL). A state mismatch or a +// memo written before the freshness bound matches nothing either — no actor, +// which is the correct answer for every human/external transition. +func (q *Queries) ConsumeStateTransition(ctx context.Context, arg ConsumeStateTransitionParams) (string, error) { + row := q.db.QueryRow(ctx, consumeStateTransition, + arg.ForgeProvider, + arg.ForgeHost, + arg.Repo, + arg.Kind, + arg.Number, + arg.State, + arg.WrittenAt, + ) + var agent_account_id string + err := row.Scan(&agent_account_id) + return agent_account_id, err +} + +const recordStateTransition = `-- name: RecordStateTransition :exec + +INSERT INTO forge_state_transitions + (forge_provider, forge_host, repo, kind, number, state, agent_account_id, written_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +ON CONFLICT (tenant_id, forge_provider, forge_host, repo, kind, number) DO UPDATE + SET state = EXCLUDED.state, + agent_account_id = EXCLUDED.agent_account_id, + written_at = EXCLUDED.written_at, + consumed_at = NULL +` + +type RecordStateTransitionParams struct { + ForgeProvider int16 + ForgeHost string + Repo string + Kind int16 + Number int64 + State string + AgentAccountID string + WrittenAt pgtype.Timestamptz +} + +// Forge state-transition memo queries (compass-forge-state-transition §Actor +// attribution). The write chokepoint upserts one memo per forge coordinate +// AFTER a successful agent-driven transition; the notify lane consumes it on +// match to attribute the echoed STATE event to the acting agent. The +// hand-written Store methods keep the door-side validation (validCoordinate), +// the state-domain guard, and the ErrInvalidArgument mapping. +// Latest transition wins: a re-transition of the same coordinate re-lands on the +// PK and RESETS consumed_at to NULL, so the newest transition is attributable +// even when the previous one was already consumed. +func (q *Queries) RecordStateTransition(ctx context.Context, arg RecordStateTransitionParams) error { + _, err := q.db.Exec(ctx, recordStateTransition, + arg.ForgeProvider, + arg.ForgeHost, + arg.Repo, + arg.Kind, + arg.Number, + arg.State, + arg.AgentAccountID, + arg.WrittenAt, + ) + return err +} diff --git a/go/internal/store/db/models.go b/go/internal/store/db/models.go index b42c365b..0f80b6e3 100644 --- a/go/internal/store/db/models.go +++ b/go/internal/store/db/models.go @@ -192,6 +192,21 @@ type ForgeRepoSubscription struct { TenantID string } +type ForgeStateTransition struct { + ForgeProvider int16 + ForgeHost string + Repo string + Kind int16 + Number int64 + State string + AgentAccountID string + WrittenAt pgtype.Timestamptz + ConsumedAt pgtype.Timestamptz + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz + TenantID string +} + type Issue struct { ID string ForgeProvider int16 diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index 35a078a6..bef83460 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -65,6 +65,12 @@ type Querier interface { ChannelsByNameForViewer(ctx context.Context, arg ChannelsByNameForViewerParams) ([]ChannelsByNameForViewerRow, error) ClearOwedMention(ctx context.Context, arg ClearOwedMentionParams) (int64, error) CollectSegment(ctx context.Context, arg CollectSegmentParams) ([]CollectSegmentRow, error) + // Single-statement clear-and-return: the row is claimed and its actor returned + // in ONE UPDATE, so a memo attributes at most one event and a concurrent second + // reader matches nothing (consumed_at is no longer NULL). A state mismatch or a + // memo written before the freshness bound matches nothing either — no actor, + // which is the correct answer for every human/external transition. + ConsumeStateTransition(ctx context.Context, arg ConsumeStateTransitionParams) (string, error) ConvertDMChannel(ctx context.Context, arg ConvertDMChannelParams) error CoordinationReports(ctx context.Context, parentAgentID pgtype.Text) ([]string, error) CountAgentForgeSubscriptionsForArtifact(ctx context.Context, arg CountAgentForgeSubscriptionsForArtifactParams) (int64, error) @@ -435,6 +441,16 @@ type Querier interface { // delivery held-deliver registry, and reaping a session that is once again live // would strand a live agent's deliveries. RecordSessionBinding(ctx context.Context, arg RecordSessionBindingParams) error + // Forge state-transition memo queries (compass-forge-state-transition §Actor + // attribution). The write chokepoint upserts one memo per forge coordinate + // AFTER a successful agent-driven transition; the notify lane consumes it on + // match to attribute the echoed STATE event to the acting agent. The + // hand-written Store methods keep the door-side validation (validCoordinate), + // the state-domain guard, and the ErrInvalidArgument mapping. + // Latest transition wins: a re-transition of the same coordinate re-lands on the + // PK and RESETS consumed_at to NULL, so the newest transition is attributable + // even when the previous one was already consumed. + RecordStateTransition(ctx context.Context, arg RecordStateTransitionParams) error RemarkSafetyValveSuperseded(ctx context.Context, arg RemarkSafetyValveSupersededParams) error RenameTopic(ctx context.Context, arg RenameTopicParams) error RequireAgentSessionSubscriber(ctx context.Context, arg RequireAgentSessionSubscriberParams) (bool, error) diff --git a/go/internal/store/forge_state_transitions.go b/go/internal/store/forge_state_transitions.go new file mode 100644 index 00000000..d2aa5115 --- /dev/null +++ b/go/internal/store/forge_state_transitions.go @@ -0,0 +1,128 @@ +package store + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgtype" + + "github.com/RigelBuild/compass/go/internal/store/db" +) + +// The agent-driven state-transition memo (design +// docs/designs/server/compass-forge-state-transition/design.md §Actor +// attribution): the durable carrier of WHICH agent drove a forge state +// transition, across the write→webhook gap. A transition has no body, so the +// DL-050 owner header cannot attribute it, and every Server-credential write +// presents the shared App bot login — so the write chokepoint records the +// acting agent here (strictly AFTER a provider success) and the notify lane +// consumes the memo on match to resolve the echoed STATE event's actor. +// +// The memo is deliberately NOT the forge_authored_artifacts row at the same +// coordinate: that row is a write-once AUTHORSHIP fact whose DO UPDATE would +// destroy the original create's F3 idempotency memo, and keying suppression off +// authorship is the author-row-proxy failure RIG-3326 rejects. + +// TransitionStateOpen and TransitionStateClosed are the portable applied-state +// domain a memo records — exactly the forge.Issue.State domain and the +// forge_state_transitions.state CHECK. A memo never carries a provider-native +// workflow-state name: the notify lane matches this value against the echoed +// event's portable state. +const ( + TransitionStateOpen = "open" + TransitionStateClosed = "closed" +) + +// validTransitionState rejects a state outside the portable domain before any DB +// round trip — the CHECK's job in Go space, so a caller bug is +// ErrInvalidArgument rather than a raw constraint violation. +func validTransitionState(state string) error { + if state != TransitionStateOpen && state != TransitionStateClosed { + return fmt.Errorf("%w: transition state must be %q or %q, got %q", ErrInvalidArgument, TransitionStateOpen, TransitionStateClosed, state) + } + return nil +} + +// RecordStateTransition upserts the memo at the forge coordinate: the portable +// state the transition APPLIED, the agent that drove it, and when it was +// written. Latest transition wins — a re-transition of the same coordinate +// re-lands on the PK and resets the consumed flag, so the newest transition is +// attributable even when the previous one was already consumed. Zero/empty +// coordinate fields, a zero kind, an out-of-domain state, an empty agent, or a +// zero timestamp -> ErrInvalidArgument; an unknown agent -> ErrInvalidArgument +// (the FK RESTRICT). +func (s *Store) RecordStateTransition(ctx context.Context, provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64, state string, agent AccountID, at time.Time) error { + if err := validCoordinate(provider, host, repo); err != nil { + return err + } + if kind == ForgeArtifactKindUnspecified { + return fmt.Errorf("%w: artifact kind is required", ErrInvalidArgument) + } + if err := validTransitionState(state); err != nil { + return err + } + if agent == "" { + return fmt.Errorf("%w: agent account id is required", ErrInvalidArgument) + } + if at.IsZero() { + return fmt.Errorf("%w: transition timestamp is required", ErrInvalidArgument) + } + if err := s.q.RecordStateTransition(ctx, db.RecordStateTransitionParams{ + ForgeProvider: int16(provider), //nolint:gosec // G115: ForgeProvider is a CHECK-constrained 1..4 enum (forge_state_transitions.forge_provider), always within int16 + ForgeHost: host, + Repo: repo, + Kind: int16(kind), //nolint:gosec // G115: ForgeArtifactKind is a CHECK-constrained 1/2 enum (forge_state_transitions.kind), always within int16 + Number: int64(number), //nolint:gosec // G115: number is a canonical forge artifact number (a positive issue/PR number) written to a BIGINT, always well within the int64 domain. + State: state, + AgentAccountID: string(agent), + WrittenAt: pgtype.Timestamptz{Time: at, Valid: true}, + }); err != nil { + if pgErrIs(err, pgForeignKeyViolation) { + return fmt.Errorf("%w: unknown agent %q", ErrInvalidArgument, agent) + } + return fmt.Errorf("store: record state transition: %w", err) + } + return nil +} + +// ConsumeStateTransition resolves and CLAIMS the memo at the forge coordinate in +// one statement: it returns the acting agent iff an unconsumed memo exists whose +// applied state matches state and whose written_at is at or after fresh (the +// freshness bound). The claim and the read are the same UPDATE … RETURNING, so +// one memo attributes at most one event and a concurrent second reader gets +// ok=false. +// +// A miss — no memo, an already-consumed memo, a state mismatch, or a memo older +// than fresh — is ok=false with NO error: that is the correct answer for every +// human/external transition and the safe answer for every race (fail-open — an +// unattributed self-transition costs one redundant wake, never a lost +// cross-agent signal). Zero/empty coordinate fields, a zero kind, or an +// out-of-domain state -> ErrInvalidArgument. +func (s *Store) ConsumeStateTransition(ctx context.Context, provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64, state string, fresh time.Time) (AccountID, bool, error) { + if err := validCoordinate(provider, host, repo); err != nil { + return "", false, err + } + if kind == ForgeArtifactKindUnspecified { + return "", false, fmt.Errorf("%w: artifact kind is required", ErrInvalidArgument) + } + if err := validTransitionState(state); err != nil { + return "", false, err + } + agent, err := s.q.ConsumeStateTransition(ctx, db.ConsumeStateTransitionParams{ + ForgeProvider: int16(provider), //nolint:gosec // G115: ForgeProvider is a CHECK-constrained 1..4 enum, always within int16 + ForgeHost: host, + Repo: repo, + Kind: int16(kind), //nolint:gosec // G115: ForgeArtifactKind is a CHECK-constrained 1/2 enum, always within int16 + Number: int64(number), //nolint:gosec // G115: number is a canonical forge artifact number written to a BIGINT, always well within the int64 domain. + State: state, + WrittenAt: pgtype.Timestamptz{Time: fresh, Valid: true}, + }) + if err != nil { + if noRows(err) { + return "", false, nil + } + return "", false, fmt.Errorf("store: consume state transition: %w", err) + } + return AccountID(agent), true, nil +} diff --git a/go/internal/store/forge_state_transitions_pgtest_test.go b/go/internal/store/forge_state_transitions_pgtest_test.go new file mode 100644 index 00000000..b5ffe5ad --- /dev/null +++ b/go/internal/store/forge_state_transitions_pgtest_test.go @@ -0,0 +1,269 @@ +//go:build pgtest + +package store + +// Real-Postgres contracts for the agent-driven state-transition memo (design +// docs/designs/server/compass-forge-state-transition/design.md §Actor +// attribution): the coordinate-keyed upsert (latest transition wins), the +// single-statement clear-and-return consume (a memo attributes at most one +// event, so a second consume finds nothing), the freshness bound, and the miss +// cases a human/external transition produces. context.Background is the test +// root (the pgtest-suite convention, sibling forge_authored_pgtest_test.go). + +import ( + "context" + "testing" + "time" +) + +// txnNow reads Postgres's own statement timestamp, so the tests' freshness +// arithmetic is anchored to the SAME clock the consume's now() stamps rather +// than to the Go process's — a skewed test box would otherwise make the +// freshness bound flaky. +func txnNow(t *testing.T, ctx context.Context, s *Store) time.Time { + t.Helper() + var now time.Time + if err := s.pool.QueryRow(ctx, `SELECT now()`).Scan(&now); err != nil { + t.Fatalf("read database now(): %v", err) + } + return now +} + +// recordAt is the common write: an issue memo at the fixture coordinate. +func recordAt(t *testing.T, ctx context.Context, s *Store, agent AccountID, state string, at time.Time) { + t.Helper() + if err := s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, state, agent, at); err != nil { + t.Fatalf("record state transition: %v", err) + } +} + +// consumeAt is the common read: consume the fixture coordinate's memo for state, +// bounded by fresh. +func consumeAt(t *testing.T, ctx context.Context, s *Store, state string, fresh time.Time) (AccountID, bool) { + t.Helper() + agent, ok, err := s.ConsumeStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, state, fresh) + if err != nil { + t.Fatalf("consume state transition: %v", err) + } + return agent, ok +} + +// ── Upsert: latest transition wins, and re-arms an already-consumed memo ────── + +func TestRecordStateTransitionUpsertLatestWins(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + first, _ := seedAgent(t, s, "fst1") + second, _ := seedAgent(t, s, "fst1-second") + now := txnNow(t, ctx, s) + + // Two transitions at the SAME coordinate: the second must REPLACE the first, + // not accrete a row — the coordinate PK is the memo's identity. + recordAt(t, ctx, s, first, TransitionStateClosed, now.Add(-2*time.Minute)) + recordAt(t, ctx, s, second, TransitionStateOpen, now.Add(-1*time.Minute)) + + var rows int + if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM forge_state_transitions`).Scan(&rows); err != nil { + t.Fatalf("count memos: %v", err) + } + if rows != 1 { + t.Fatalf("memo rows = %d, want 1 (upsert on the coordinate PK, not a second row)", rows) + } + + // The superseded state no longer matches; the latest one resolves its agent. + if agent, ok := consumeAt(t, ctx, s, TransitionStateClosed, now.Add(-time.Hour)); ok { + t.Fatalf("superseded state resolved agent %q, want no match", agent) + } + agent, ok := consumeAt(t, ctx, s, TransitionStateOpen, now.Add(-time.Hour)) + if !ok || agent != second { + t.Fatalf("latest transition resolved (%q, %v), want (%q, true)", agent, ok, second) + } +} + +// A re-transition must RE-ARM the memo: the upsert clears consumed_at, so the +// newest transition is attributable even though the previous one at the same +// coordinate was already consumed. Without the reset, an agent could close an +// issue, have that event attributed, then reopen it and go unattributed forever. +func TestRecordStateTransitionUpsertReArmsConsumedMemo(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + agent, _ := seedAgent(t, s, "fst2") + now := txnNow(t, ctx, s) + + recordAt(t, ctx, s, agent, TransitionStateClosed, now.Add(-2*time.Minute)) + if _, ok := consumeAt(t, ctx, s, TransitionStateClosed, now.Add(-time.Hour)); !ok { + t.Fatalf("first consume found no memo") + } + + recordAt(t, ctx, s, agent, TransitionStateOpen, now.Add(-time.Minute)) + got, ok := consumeAt(t, ctx, s, TransitionStateOpen, now.Add(-time.Hour)) + if !ok || got != agent { + t.Fatalf("re-armed memo resolved (%q, %v), want (%q, true)", got, ok, agent) + } +} + +// ── Consume-once: one memo attributes at most one event ────────────────────── + +func TestConsumeStateTransitionConsumesExactlyOnce(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + agent, _ := seedAgent(t, s, "fst3") + now := txnNow(t, ctx, s) + fresh := now.Add(-time.Hour) + + recordAt(t, ctx, s, agent, TransitionStateClosed, now.Add(-time.Minute)) + + got, ok := consumeAt(t, ctx, s, TransitionStateClosed, fresh) + if !ok || got != agent { + t.Fatalf("first consume = (%q, %v), want (%q, true)", got, ok, agent) + } + // The clear and the return are ONE statement, so the memo is already claimed: + // a second reader (the reconcile sweep racing the webhook arm) resolves no + // actor and its event is delivered unattributed — the documented fail-open. + if got, ok := consumeAt(t, ctx, s, TransitionStateClosed, fresh); ok { + t.Fatalf("second consume = (%q, true), want no actor (the memo was already claimed)", got) + } + + // The row survives its consume (stamped, not deleted), so the claim is + // durable rather than depending on a delete the next upsert would race. + var consumed bool + if err := s.pool.QueryRow(ctx, `SELECT consumed_at IS NOT NULL FROM forge_state_transitions`).Scan(&consumed); err != nil { + t.Fatalf("read consumed_at: %v", err) + } + if !consumed { + t.Fatalf("consumed_at is still NULL after a successful consume") + } +} + +// ── The freshness bound ─────────────────────────────────────────────────────── + +func TestConsumeStateTransitionFreshnessBound(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + agent, _ := seedAgent(t, s, "fst4") + now := txnNow(t, ctx, s) + + // A memo written BEFORE the bound is stale: a lingering memo (the Route + // cursor-advance race) must not attribute an event minutes later. + recordAt(t, ctx, s, agent, TransitionStateClosed, now.Add(-10*time.Minute)) + if got, ok := consumeAt(t, ctx, s, TransitionStateClosed, now.Add(-5*time.Minute)); ok { + t.Fatalf("stale memo resolved agent %q, want no match", got) + } + // Rejecting a stale memo must NOT consume it — the bound is a filter, not a + // sweep. Relaxing the bound still finds it. + got, ok := consumeAt(t, ctx, s, TransitionStateClosed, now.Add(-time.Hour)) + if !ok || got != agent { + t.Fatalf("relaxed bound = (%q, %v), want (%q, true) — the stale read must not have consumed the memo", got, ok, agent) + } +} + +// The bound is inclusive at its edge: a memo written exactly AT fresh resolves. +// An exclusive comparison would drop an on-the-boundary memo, so the edge is +// pinned rather than left to the operator's reading of ">=". +func TestConsumeStateTransitionFreshnessBoundIsInclusive(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + agent, _ := seedAgent(t, s, "fst5") + at := txnNow(t, ctx, s).Add(-time.Minute) + + recordAt(t, ctx, s, agent, TransitionStateClosed, at) + got, ok := consumeAt(t, ctx, s, TransitionStateClosed, at) + if !ok || got != agent { + t.Fatalf("memo written exactly at the bound = (%q, %v), want (%q, true)", got, ok, agent) + } +} + +// ── Miss cases: every human/external transition lands here ─────────────────── + +func TestConsumeStateTransitionMisses(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + agent, _ := seedAgent(t, s, "fst6") + now := txnNow(t, ctx, s) + fresh := now.Add(-time.Hour) + + // No memo at all — a human closing an issue Compass never transitioned. + if got, ok := consumeAt(t, ctx, s, TransitionStateClosed, fresh); ok { + t.Fatalf("no-memo consume = (%q, true), want no actor", got) + } + + recordAt(t, ctx, s, agent, TransitionStateClosed, now.Add(-time.Minute)) + + // State mismatch: the echoed event's state is not the one the agent applied, + // so the memo does not attribute it (and is NOT consumed by the near-miss). + if got, ok := consumeAt(t, ctx, s, TransitionStateOpen, fresh); ok { + t.Fatalf("state-mismatch consume = (%q, true), want no actor", got) + } + + // Coordinate mismatch on each key component: a memo attributes ONLY its own + // artifact. Sharing a repo, a number, or a kind must not be enough. + for _, m := range []struct { + name string + provider ForgeProvider + host string + repo string + kind ForgeArtifactKind + number uint64 + }{ + {"other provider", ForgeProviderLinear, "github.com", "a/b", ForgeArtifactKindIssue, 42}, + {"other host", ForgeProviderGitHub, "ghe.example", "a/b", ForgeArtifactKindIssue, 42}, + {"other repo", ForgeProviderGitHub, "github.com", "a/c", ForgeArtifactKindIssue, 42}, + {"other kind", ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindPullRequest, 42}, + {"other number", ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 43}, + } { + t.Run(m.name, func(t *testing.T) { + got, ok, err := s.ConsumeStateTransition(ctx, m.provider, m.host, m.repo, m.kind, m.number, TransitionStateClosed, fresh) + if err != nil { + t.Fatalf("consume: %v", err) + } + if ok { + t.Fatalf("%s resolved agent %q, want no actor", m.name, got) + } + }) + } + + // None of the misses consumed the real memo: its own coordinate still + // resolves. A miss that silently claimed the row would lose the attribution + // the memo exists for. + got, ok := consumeAt(t, ctx, s, TransitionStateClosed, fresh) + if !ok || got != agent { + t.Fatalf("exact match after the misses = (%q, %v), want (%q, true)", got, ok, agent) + } +} + +// ── Door-side validation: a caller bug never reaches Postgres ──────────────── + +func TestStateTransitionValidation(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + agent, _ := seedAgent(t, s, "fst7") + at := txnNow(t, ctx, s) + + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderUnspecified, "github.com", "a/b", ForgeArtifactKindIssue, 42, TransitionStateClosed, agent, at), + ErrInvalidArgument, "record with an unspecified provider") + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "", "a/b", ForgeArtifactKindIssue, 42, TransitionStateClosed, agent, at), + ErrInvalidArgument, "record with an empty host") + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "", ForgeArtifactKindIssue, 42, TransitionStateClosed, agent, at), + ErrInvalidArgument, "record with an empty repo") + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindUnspecified, 42, TransitionStateClosed, agent, at), + ErrInvalidArgument, "record with an unspecified kind") + // The state domain is guarded in Go space, so an out-of-domain value is a + // caller bug rather than a raw CHECK violation surfacing from the driver. + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, "merged", agent, at), + ErrInvalidArgument, "record with an out-of-domain state") + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, TransitionStateClosed, "", at), + ErrInvalidArgument, "record with an empty agent") + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, TransitionStateClosed, agent, time.Time{}), + ErrInvalidArgument, "record with a zero timestamp") + // The FK RESTRICT: a memo can only name a real agent, so a deleted or + // fabricated account cannot be recorded as an actor. + sentinelIs(t, s.RecordStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, TransitionStateClosed, "no-such-agent", at), + ErrInvalidArgument, "record with an unknown agent") + + _, _, err := s.ConsumeStateTransition(ctx, ForgeProviderUnspecified, "github.com", "a/b", ForgeArtifactKindIssue, 42, TransitionStateClosed, at) + sentinelIs(t, err, ErrInvalidArgument, "consume with an unspecified provider") + _, _, err = s.ConsumeStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindUnspecified, 42, TransitionStateClosed, at) + sentinelIs(t, err, ErrInvalidArgument, "consume with an unspecified kind") + _, _, err = s.ConsumeStateTransition(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 42, "merged", at) + sentinelIs(t, err, ErrInvalidArgument, "consume with an out-of-domain state") +} diff --git a/go/internal/store/migrations/0001_init.sql b/go/internal/store/migrations/0001_init.sql index 0b4f7867..860f81a4 100644 --- a/go/internal/store/migrations/0001_init.sql +++ b/go/internal/store/migrations/0001_init.sql @@ -912,6 +912,61 @@ CREATE UNIQUE INDEX forge_authored_artifacts_request_memo_idx CREATE INDEX forge_authored_artifacts_agent_idx ON forge_authored_artifacts (agent_account_id); +-- One row per forge coordinate an AGENT-DRIVEN state transition last landed on +-- (compass-forge-state-transition design.md §Actor attribution): the consumable +-- memo that carries the acting agent's identity across the write→webhook gap. +-- A transition has no body, so the DL-050 owner header cannot attribute it, and +-- every Server-credential write presents the shared App bot login — durable +-- server-side correlation is the only channel that can name WHICH agent drove +-- the transition. The write chokepoint upserts this row strictly AFTER a +-- provider success (a rejected transition leaves no memo); the notify lane +-- resolves a STATE event's actor by consuming it. +-- +-- Coordinate-aligned to forge_authored_artifacts: the SAME (tenant_id, +-- forge_provider, forge_host, repo, kind, number) PK, so a re-transition of one +-- artifact re-lands on the key (latest transition wins) rather than accreting +-- rows. This table is deliberately NOT forge_authored_artifacts itself: that +-- row is a write-once AUTHORSHIP fact whose DO UPDATE would destroy the +-- original create's F3 idempotency memo. +-- +-- tenant_id is load-bearing, not incidental: two tenants legitimately hold the +-- SAME forge coordinate (TestForgeAuthoredTwoTenantsSameCoordinate), so without +-- it one tenant's memo could attribute another tenant's STATE event. +-- +-- state is the APPLIED PORTABLE target, matched against the echoed event's +-- state — never a provider-native workflow-state name, hence CHECK IN +-- ('open', 'closed'). consumed_at NULL means "unconsumed"; the consume is a +-- single UPDATE … RETURNING that stamps it, so one memo attributes at most one +-- event and a concurrent second reader matches nothing. written_at is the +-- freshness anchor: a memo older than the reader's bound resolves no actor +-- (fail-open — an unattributed self-transition costs one redundant wake, never +-- a lost cross-agent signal). written_at is NOT this row's local mutation time +-- and is deliberately distinct from the created_at/updated_at pair below: it is +-- the chokepoint-supplied anchor ConsumeStateTransition compares against its +-- freshness bound, so it is set explicitly by the writer, never by the trigger. +CREATE TABLE forge_state_transitions ( + forge_provider SMALLINT NOT NULL CHECK (forge_provider IN (1, 2, 3, 4)), + forge_host TEXT NOT NULL, + repo TEXT NOT NULL, + kind SMALLINT NOT NULL CHECK (kind IN (1, 2)), + number BIGINT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('open', 'closed')), + -- SINGLE-column FK, deliberately diverging from the composite + -- (agent_account_id, owner_user_id) FK the otherwise field-for-field + -- sibling forge_authored_artifacts carries: that row records an + -- AUTHORSHIP pair whose (agent, that-agent's-owner) halves must be + -- validated together, while this memo records only the ACTING agent, so + -- there is no pair to validate. account_id is agent_accounts' PK, hence + -- globally unique, so the single-column reference is fully constrained. + agent_account_id TEXT NOT NULL REFERENCES agent_accounts (account_id) ON DELETE RESTRICT, + written_at TIMESTAMPTZ NOT NULL, + consumed_at TIMESTAMPTZ, -- NULL = unconsumed; stamped by the one-shot consume + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + tenant_id TEXT NOT NULL DEFAULT current_setting('compass.tenant_id', TRUE), + PRIMARY KEY (tenant_id, forge_provider, forge_host, repo, kind, number) +); + -- linear_agent_sessions: the Linear Agent Session ↔ Compass conversation -- association (compass-linear-agent-responder design.md §Part 2 / §T3). One row -- per Linear AgentSession the responder has handled: the resolved Manager, that @@ -1030,7 +1085,8 @@ DECLARE 'agent_delivery_cursors', 'owed_mentions', 'agent_activity', 'agent_forge_subscriptions', 'forge_authored_artifacts', 'linear_agent_sessions', - 'issues', 'forge_repo_subscriptions', 'forge_artifact_cursors' + 'issues', 'forge_repo_subscriptions', 'forge_artifact_cursors', + 'forge_state_transitions' ]; BEGIN FOREACH t IN ARRAY tenant_tables LOOP @@ -1109,7 +1165,8 @@ DECLARE 'session_bindings', 'agent_config_bundle', 'model_registry', - 'forge_repo_subscriptions' + 'forge_repo_subscriptions', + 'forge_state_transitions' ]; BEGIN FOREACH t IN ARRAY updated_at_tables LOOP diff --git a/go/internal/store/queries/forge_state_transitions.sql b/go/internal/store/queries/forge_state_transitions.sql new file mode 100644 index 00000000..65638a4d --- /dev/null +++ b/go/internal/store/queries/forge_state_transitions.sql @@ -0,0 +1,37 @@ +-- Forge state-transition memo queries (compass-forge-state-transition §Actor +-- attribution). The write chokepoint upserts one memo per forge coordinate +-- AFTER a successful agent-driven transition; the notify lane consumes it on +-- match to attribute the echoed STATE event to the acting agent. The +-- hand-written Store methods keep the door-side validation (validCoordinate), +-- the state-domain guard, and the ErrInvalidArgument mapping. + +-- name: RecordStateTransition :exec +-- Latest transition wins: a re-transition of the same coordinate re-lands on the +-- PK and RESETS consumed_at to NULL, so the newest transition is attributable +-- even when the previous one was already consumed. +INSERT INTO forge_state_transitions + (forge_provider, forge_host, repo, kind, number, state, agent_account_id, written_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +ON CONFLICT (tenant_id, forge_provider, forge_host, repo, kind, number) DO UPDATE + SET state = EXCLUDED.state, + agent_account_id = EXCLUDED.agent_account_id, + written_at = EXCLUDED.written_at, + consumed_at = NULL; + +-- name: ConsumeStateTransition :one +-- Single-statement clear-and-return: the row is claimed and its actor returned +-- in ONE UPDATE, so a memo attributes at most one event and a concurrent second +-- reader matches nothing (consumed_at is no longer NULL). A state mismatch or a +-- memo written before the freshness bound matches nothing either — no actor, +-- which is the correct answer for every human/external transition. +UPDATE forge_state_transitions + SET consumed_at = now() + WHERE forge_provider = $1 + AND forge_host = $2 + AND repo = $3 + AND kind = $4 + AND number = $5 + AND state = $6 + AND consumed_at IS NULL + AND written_at >= $7 +RETURNING agent_account_id; diff --git a/go/internal/store/rls_pgtest_test.go b/go/internal/store/rls_pgtest_test.go index 1b29ed82..bdd2f648 100644 --- a/go/internal/store/rls_pgtest_test.go +++ b/go/internal/store/rls_pgtest_test.go @@ -598,6 +598,7 @@ func TestRLSCatalogEnabledAndForced(t *testing.T) { "agent_forge_subscriptions", "forge_authored_artifacts", "linear_agent_sessions", "issues", "forge_repo_subscriptions", "forge_artifact_cursors", + "forge_state_transitions", } for _, tbl := range tenantOwned { if !enumerated[tbl] { diff --git a/go/server/forge.go b/go/server/forge.go index 4fa3471c..065e649e 100644 --- a/go/server/forge.go +++ b/go/server/forge.go @@ -133,16 +133,18 @@ func (r *forgeProviderRegistry) resolve(ref *compassv1.ForgeRef) (resolvedForge, // forgeStore is the narrow store surface the chokepoint needs: resolve the // caller's attribution (agent handle + owning user), the F3 idempotency-memo -// lookup, and the DL-055 ownership-row + memo write. Satisfied by *store.Store; -// a narrow interface (the issueStore / CommsCaller pattern) so the -// stamp/dedup/record ordering is provable in the default lane against a fake, -// not only behind the pgtest tag. +// lookup, the DL-055 ownership-row + memo write, and the state-transition memo +// write. Satisfied by *store.Store; a narrow interface (the issueStore / +// CommsCaller pattern) so the stamp/dedup/record ordering — and the transition +// arms' memo-strictly-after-provider-success ordering — is provable in the +// default lane against a fake, not only behind the pgtest tag. type forgeStore interface { GetAccount(ctx context.Context, id store.AccountID) (store.Account, error) AuthoredArtifactByRequestID(ctx context.Context, agent store.AccountID, clientRequestID string) (store.AuthoredArtifact, bool, error) RecordAuthoredArtifact(ctx context.Context, a store.AuthoredArtifact) error EnsureAgentForgeSubscription(ctx context.Context, sub store.AgentForgeSubscription) (string, error) DeleteAgentForgeSubscription(ctx context.Context, agent store.AccountID, subscriptionID string) error + RecordStateTransition(ctx context.Context, provider store.ForgeProvider, host, repo string, kind store.ForgeArtifactKind, number uint64, state string, agent store.AccountID, at time.Time) error } // forgeService is the ForgeCaller implementation and the DL-050 write @@ -202,6 +204,10 @@ func (s *forgeService) ExecuteForgeCallAsAccount( return s.commentOnPullRequest(ctx, caller, sessionID, call, c.CommentOnPullRequest), nil case *compassv1internal.ForgeCallRequest_SubmitReview: return s.submitReview(ctx, caller, sessionID, call, c.SubmitReview), nil + case *compassv1internal.ForgeCallRequest_TransitionIssueState: + return s.transitionIssueState(ctx, caller, call, c.TransitionIssueState), nil + case *compassv1internal.ForgeCallRequest_TransitionPullRequestState: + return s.transitionPullRequestState(ctx, caller, call, c.TransitionPullRequestState), nil case *compassv1internal.ForgeCallRequest_GetIssue: return s.getIssue(ctx, call, c.GetIssue), nil case *compassv1internal.ForgeCallRequest_ListIssues: @@ -530,6 +536,137 @@ func (s *forgeService) submitReview(ctx context.Context, caller store.AccountID, } } +// transitionStateDomain is the portable target-state domain both transition arms +// screen against — exactly forge.Issue.State's, and exactly the store memo's. +// Anything else is an in-band invalid_argument BEFORE any provider touch. +func transitionStateDomain(state string) *compassv1internal.ForgeCallError { + if state != store.TransitionStateOpen && state != store.TransitionStateClosed { + return forgeErr(connect.CodeInvalidArgument, fmt.Sprintf( + "forge: state must be %q or %q, got %q", store.TransitionStateOpen, store.TransitionStateClosed, state)) + } + return nil +} + +// screenIssueRefinements is the refinement/provider screen: close_reason is a +// GitHub-ISSUE concept and workflow_state is a Linear one, so a refinement the +// ADDRESSED provider cannot express is rejected HERE, in-band, before any +// provider call. Silently dropping it is rejected — the caller asked for +// something specific and the write would not do it. Screening at the arm is +// also what lets forge.TransitionState document that a provider may safely +// ignore a foreign field. +func screenIssueRefinements(rf resolvedForge, closeReason, workflowState string) *compassv1internal.ForgeCallError { + if closeReason != "" && rf.provider != compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB { + return forgeErr(connect.CodeInvalidArgument, fmt.Sprintf( + "forge: close_reason is a GitHub-issue refinement and provider %q cannot express it", rf.author.Name())) + } + if workflowState != "" && rf.provider != compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR { + return forgeErr(connect.CodeInvalidArgument, fmt.Sprintf( + "forge: workflow_state is a Linear refinement and provider %q cannot express it", rf.author.Name())) + } + return nil +} + +// rememberTransition writes the actor memo (§Actor attribution) at the +// transitioned coordinate, STRICTLY AFTER the provider succeeded — so a rejected +// transition leaves no memo, mirroring record's "strictly AFTER a create's forge +// success" ordering. The memo carries the REQUESTED portable state, not the +// returned artifact's raw one: the notify lane matches it against the echoed +// STATE event's portable state, and a raw PR state can be "merged", outside that +// domain. There is no DL-055 row and no F3 memo here: a transition mints no +// coordinate, and the authored row at this coordinate is a write-once authorship +// fact whose upsert would destroy the original create's idempotency memo. +// +// COORDINATE CONTRACT (load-bearing, unguarded): the memo is keyed on the +// CALLER-supplied repo string plus the PROVIDER-returned number, while the +// notify lane's ConsumeStateTransition keys on the coordinate the WEBHOOK +// carries — wh.Repository.FullName on GitHub (canonical "owner/name", +// githubapp_webhook.go) and de.Data.Team.Key on Linear (canonical team key, +// linearagent/data_event.go). So req.GetRepo() MUST already equal that +// canonical form for the memo to resolve. A caller addressing the same repo in +// different casing (or any other non-canonical spelling) writes a memo the +// webhook can never match, and the transition goes UNATTRIBUTED — the +// documented fail-open (one redundant wake, never a lost cross-agent signal), +// so the divergence is SILENT: nothing here normalizes, guards, or asserts it. +// Normalizing would change the coordinate's canonical form, which is a design +// decision, not a fix applied here. +func (s *forgeService) rememberTransition(ctx context.Context, rf resolvedForge, caller store.AccountID, repo string, kind store.ForgeArtifactKind, number uint64, state string) *compassv1internal.ForgeCallError { + if err := s.store.RecordStateTransition(ctx, store.ForgeProvider(rf.provider), rf.host, repo, kind, number, state, caller, s.now()); err != nil { + // The provider write ALREADY LANDED, so the error must name what + // succeeded: a caller told only "memo: db unavailable" reasonably + // retries an operation that is already done, and a human reading the + // trace cannot tell the forge state changed. The mapped code is + // unchanged — only the message gains the landed half. + fe := storeForgeError(err) + fe.Message = fmt.Sprintf( + "forge: %s#%d was transitioned to %s, but recording the acting agent failed: %v", + repo, number, state, err) + return fe + } + return nil +} + +// transitionIssueState moves an existing issue between forge states: resolve +// target, screen the state domain and the refinement/provider match, dispatch on +// the AUTHOR client, flatten, write the actor memo on success, and return the +// UPDATED issue on the EXISTING Issue result arm — the caller sees +// post-transition truth the same way a create's caller sees the created +// artifact. No stamp (a transition has no body) and no F3 dedup +// (client_request_id is ignored on non-create arms; a retried transition +// converges on the target state rather than duplicating anything). +func (s *forgeService) transitionIssueState(ctx context.Context, caller store.AccountID, call *compassv1internal.ForgeCallRequest, req *compassv1internal.TransitionIssueStateRequest) *compassv1internal.ForgeCallResult { + rf, fe := s.resolveTarget(call, req.GetRepo()) + if fe != nil { + return forgeErrorResult(fe) + } + if fe := transitionStateDomain(req.GetState()); fe != nil { + return forgeErrorResult(fe) + } + if fe := screenIssueRefinements(rf, req.GetCloseReason(), req.GetWorkflowState()); fe != nil { + return forgeErrorResult(fe) + } + iss, err := rf.author.TransitionIssueState(ctx, req.GetRepo(), req.GetIssueNumber(), forge.TransitionState{ + State: req.GetState(), + CloseReason: req.GetCloseReason(), + WorkflowState: req.GetWorkflowState(), + }) + if err != nil { + return forgeErrorResult(mapForgeError(err, forgeOp{provider: rf.author.Name(), op: "transition_issue_state"})) + } + if fe := s.rememberTransition(ctx, rf, caller, req.GetRepo(), store.ForgeArtifactKindIssue, iss.Number, req.GetState()); fe != nil { + return forgeErrorResult(fe) + } + return &compassv1internal.ForgeCallResult{ + Result: &compassv1internal.ForgeCallResult_Issue{Issue: translateIssue(iss, rf, req.GetRepo())}, + } +} + +// transitionPullRequestState is the PR twin of transitionIssueState. The PR arm +// accepts NO refinement fields at all — close_reason is a GitHub *issue* +// concept and workflow_state a Linear one, and merge is a separate concern never +// expressed as a transition — so the wire message carries none and the dispatched +// forge.TransitionState is state-only. A provider with no PR model answers +// ErrUnsupported, which mapForgeError flattens to unimplemented naming +// provider+op. +func (s *forgeService) transitionPullRequestState(ctx context.Context, caller store.AccountID, call *compassv1internal.ForgeCallRequest, req *compassv1internal.TransitionPullRequestStateRequest) *compassv1internal.ForgeCallResult { + rf, fe := s.resolveTarget(call, req.GetRepo()) + if fe != nil { + return forgeErrorResult(fe) + } + if fe := transitionStateDomain(req.GetState()); fe != nil { + return forgeErrorResult(fe) + } + pr, err := rf.author.TransitionPullRequestState(ctx, req.GetRepo(), req.GetPrNumber(), forge.TransitionState{State: req.GetState()}) + if err != nil { + return forgeErrorResult(mapForgeError(err, forgeOp{provider: rf.author.Name(), op: "transition_pull_request_state"})) + } + if fe := s.rememberTransition(ctx, rf, caller, req.GetRepo(), store.ForgeArtifactKindPullRequest, pr.Number, req.GetState()); fe != nil { + return forgeErrorResult(fe) + } + return &compassv1internal.ForgeCallResult{ + Result: &compassv1internal.ForgeCallResult_PullRequest{PullRequest: translatePR(pr, rf, req.GetRepo())}, + } +} + // getIssue answers from the issue projection for a TRACKED artifact (OQ-A), else // composes a live author-client fetch + TranslateIssue. Read bodies are stripped // of the owner header (never stamped) — provider truth, not attribution. diff --git a/go/server/forge_test.go b/go/server/forge_test.go index 4c0eaff4..38fabc6a 100644 --- a/go/server/forge_test.go +++ b/go/server/forge_test.go @@ -54,6 +54,12 @@ type fakeForgeStore struct { getErr error // if set, GetAccount returns it verbatim recErr error // if set, RecordAuthoredArtifact returns it verbatim + // The state-transition actor memo: transitions records every + // RecordStateTransition in call order (so the ordering against the provider + // log is assertable), and transErr forces a memo-write fault. + transitions []recordedTransition + transErr error + // DL-053 subscriptions: subs is keyed by subscription id; subKey indexes the // UNIQUE (agent, coordinate) to the existing id so a repeat subscribe is // idempotent, exactly as the real store's ON CONFLICT does. subErr / delErr @@ -65,6 +71,20 @@ type fakeForgeStore struct { delErr error } +// recordedTransition is one RecordStateTransition the fake saw: the full +// argument list, so a test can assert the coordinate, the APPLIED portable +// state, the acting agent, and the clock the chokepoint stamped. +type recordedTransition struct { + provider store.ForgeProvider + host string + repo string + kind store.ForgeArtifactKind + number uint64 + state string + agent store.AccountID + at time.Time +} + func newFakeForgeStore() *fakeForgeStore { return &fakeForgeStore{ accounts: make(map[store.AccountID]store.Account), @@ -143,6 +163,20 @@ func (f *fakeForgeStore) RecordAuthoredArtifact(_ context.Context, a store.Autho return nil } +// RecordStateTransition mirrors the real store's upsert-latest-wins memo write: +// it appends to an ordered log so a test can prove the memo landed STRICTLY +// AFTER the provider call (and never at all when the provider failed). +func (f *fakeForgeStore) RecordStateTransition(_ context.Context, provider store.ForgeProvider, host, repo string, kind store.ForgeArtifactKind, number uint64, state string, agent store.AccountID, at time.Time) error { + if f.transErr != nil { + return f.transErr + } + f.transitions = append(f.transitions, recordedTransition{ + provider: provider, host: host, repo: repo, kind: kind, + number: number, state: state, agent: agent, at: at, + }) + return nil +} + // seedAgent registers an agent account and its owning user so resolveIdentity // finds both handles. func (f *fakeForgeStore) seedAgent(agentID store.AccountID, agentHandle string, ownerID store.AccountID, ownerHandle string) { diff --git a/go/server/forge_transition_test.go b/go/server/forge_transition_test.go new file mode 100644 index 00000000..ec23e1bd --- /dev/null +++ b/go/server/forge_transition_test.go @@ -0,0 +1,433 @@ +//go:build unix + +package server + +// Default-lane (no database) tests for the two forge state-transition arms +// (compass-forge-state-transition design.md §The server arm / §Actor +// attribution). They ride the same harness as forge_test.go — the exported +// forge.FakeProvider plus the faithful in-memory forgeStore — so the arm +// pipeline (resolveTarget → state-domain screen → refinement/provider screen → +// author-client dispatch → mapForgeError → actor memo → updated artifact) is +// observable end to end without Postgres. The memo's real-Postgres contract +// (upsert-latest-wins, consume-once, freshness) is the store package's own +// pgtest suite (forge_state_transitions_pgtest_test.go). + +import ( + "errors" + "strings" + "testing" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/forge" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/store" +) + +// transitionIssueCall builds a TransitionIssueState request against the default +// coordinate, carrying whichever refinements the screen under test needs. +func transitionIssueCall(state, closeReason, workflowState string) *compassv1internal.ForgeCallRequest { + return &compassv1internal.ForgeCallRequest{ + Call: &compassv1internal.ForgeCallRequest_TransitionIssueState{TransitionIssueState: &compassv1internal.TransitionIssueStateRequest{ + Repo: testRepo, IssueNumber: 7, State: state, CloseReason: closeReason, WorkflowState: workflowState, + }}, + } +} + +// transitionPRCall builds a TransitionPullRequestState request against the +// default coordinate. The PR arm carries no refinement fields at all. +func transitionPRCall(state string) *compassv1internal.ForgeCallRequest { + return &compassv1internal.ForgeCallRequest{ + Call: &compassv1internal.ForgeCallRequest_TransitionPullRequestState{TransitionPullRequestState: &compassv1internal.TransitionPullRequestStateRequest{ + Repo: testRepo, PrNumber: 9, State: state, + }}, + } +} + +// newLinearForgeServiceForTest builds a service whose DEFAULT coordinate is +// LINEAR, so the refinement screen's provider-specific half is drivable in both +// directions (close_reason rejected on Linear, workflow_state rejected on +// GitHub) rather than only the GitHub one. +func newLinearForgeServiceForTest(t *testing.T, author *forge.FakeProvider) (*forgeService, *fakeForgeStore) { + t.Helper() + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("lin-reviewer")) + reg := newForgeProviderRegistry() + reg.register(forgeCoordinate{provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, host: "linear.app"}, author, author, true) + svc.providers = reg + return svc, st +} + +// --- tests: arm dispatch ------------------------------------------------------ + +// TestForgeTransitionIssueStateDispatchesAndReturnsUpdatedIssue pins the issue +// arm end to end: the AUTHOR client's TransitionIssueState is invoked with the +// requested coordinate and portable target, and the result comes back on the +// EXISTING Issue arm carrying the UPDATED artifact (post-transition truth), not +// a new result arm and not the pre-transition state. +func TestForgeTransitionIssueStateDispatchesAndReturnsUpdatedIssue(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + reviewer := forge.NewFakeProvider("gh-reviewer") + svc, _ := newForgeServiceForTest(t, author, reviewer) + author.TransitionIssueResult = forge.Issue{Number: 7, Title: "t", State: "closed"} + + res := svc.ExecuteForgeCallAsAccountMust(t, transitionIssueCall("closed", "completed", "")) + if fe := res.GetError(); fe != nil { + t.Fatalf("transition returned an in-band error: %v", fe) + } + iss := res.GetIssue() + if iss == nil { + t.Fatalf("result arm = %T, want the Issue arm", res.GetResult()) + } + if iss.GetNumber() != 7 || iss.GetRepo() != testRepo { + t.Fatalf("issue coordinate = %s#%d, want %s#7", iss.GetRepo(), iss.GetNumber(), testRepo) + } + if iss.GetForge().GetProvider() != compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB || iss.GetForge().GetHost() != testHost { + t.Fatalf("issue forge ref = %v, want the resolved coordinate", iss.GetForge()) + } + + calls := author.Calls() + if len(calls) != 1 || calls[0].Method != "TransitionIssueState" { + t.Fatalf("author calls = %+v, want one TransitionIssueState", calls) + } + if calls[0].Repo != testRepo || calls[0].Number != 7 { + t.Fatalf("dispatched coordinate = %s#%d, want %s#7", calls[0].Repo, calls[0].Number, testRepo) + } + in, ok := calls[0].Payload.(forge.TransitionState) + if !ok { + t.Fatalf("payload = %T, want forge.TransitionState", calls[0].Payload) + } + if in.State != "closed" || in.CloseReason != "completed" || in.WorkflowState != "" { + t.Fatalf("dispatched input = %+v, want {closed completed }", in) + } + if len(reviewer.Calls()) != 0 { + t.Fatalf("reviewer calls = %d, want 0 (a transition is an ordinary author write)", len(reviewer.Calls())) + } +} + +// TestForgeTransitionPullRequestStateDispatchesAndReturnsUpdatedPR is the PR +// twin: the AUTHOR client's TransitionPullRequestState runs with a state-ONLY +// input (the PR arm has no refinement fields) and the updated PR comes back on +// the existing PullRequest arm. +func TestForgeTransitionPullRequestStateDispatchesAndReturnsUpdatedPR(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + reviewer := forge.NewFakeProvider("gh-reviewer") + svc, _ := newForgeServiceForTest(t, author, reviewer) + author.TransitionPRResult = forge.PullRequest{Number: 9, Title: "t", State: "closed"} + + res := svc.ExecuteForgeCallAsAccountMust(t, transitionPRCall("closed")) + if fe := res.GetError(); fe != nil { + t.Fatalf("transition returned an in-band error: %v", fe) + } + pr := res.GetPullRequest() + if pr == nil { + t.Fatalf("result arm = %T, want the PullRequest arm", res.GetResult()) + } + if pr.GetNumber() != 9 || pr.GetRepo() != testRepo { + t.Fatalf("PR coordinate = %s#%d, want %s#9", pr.GetRepo(), pr.GetNumber(), testRepo) + } + + calls := author.Calls() + if len(calls) != 1 || calls[0].Method != "TransitionPullRequestState" { + t.Fatalf("author calls = %+v, want one TransitionPullRequestState", calls) + } + in, ok := calls[0].Payload.(forge.TransitionState) + if !ok { + t.Fatalf("payload = %T, want forge.TransitionState", calls[0].Payload) + } + if in != (forge.TransitionState{State: "closed"}) { + t.Fatalf("dispatched input = %+v, want a state-only {closed}", in) + } +} + +// --- tests: validation screens ------------------------------------------------ + +// TestForgeTransitionStateOutsideDomainIsInvalidArgument pins the state-domain +// screen on BOTH arms: only "open" and "closed" are portable targets, so a +// provider-native name ("merged", a Linear column name) or an empty state is an +// in-band invalid_argument with ZERO provider calls and ZERO memo writes. +func TestForgeTransitionStateOutsideDomainIsInvalidArgument(t *testing.T) { + for _, state := range []string{"", "merged", "Done", "OPEN"} { + for _, arm := range []struct { + name string + call func(string) *compassv1internal.ForgeCallRequest + }{ + {"issue", func(s string) *compassv1internal.ForgeCallRequest { return transitionIssueCall(s, "", "") }}, + {"pull_request", transitionPRCall}, + } { + t.Run(arm.name+"/"+state, func(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + + fe := svc.ExecuteForgeCallAsAccountMust(t, arm.call(state)).GetError() + if fe == nil || fe.GetCode() != "invalid_argument" { + t.Fatalf("state %q error = %v, want invalid_argument", state, fe) + } + if len(author.Calls()) != 0 || len(st.transitions) != 0 { + t.Fatalf("rejected state touched provider/store: provider=%d memo=%d", len(author.Calls()), len(st.transitions)) + } + }) + } + } +} + +// TestForgeTransitionEmptyRepoIsInvalidArgument pins that the transition arms +// inherit resolveTarget's posture: an empty repo is invalid_argument before any +// store or provider touch, exactly as on every other call. +func TestForgeTransitionEmptyRepoIsInvalidArgument(t *testing.T) { + calls := map[string]*compassv1internal.ForgeCallRequest{ + "issue": {Call: &compassv1internal.ForgeCallRequest_TransitionIssueState{TransitionIssueState: &compassv1internal.TransitionIssueStateRequest{ + Repo: "", IssueNumber: 7, State: "closed", + }}}, + "pull_request": {Call: &compassv1internal.ForgeCallRequest_TransitionPullRequestState{TransitionPullRequestState: &compassv1internal.TransitionPullRequestStateRequest{ + Repo: "", PrNumber: 9, State: "closed", + }}}, + } + for name, call := range calls { + t.Run(name, func(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + + fe := svc.ExecuteForgeCallAsAccountMust(t, call).GetError() + if fe == nil || fe.GetCode() != "invalid_argument" { + t.Fatalf("empty repo error = %v, want invalid_argument", fe) + } + if len(author.Calls()) != 0 || len(st.transitions) != 0 { + t.Fatalf("empty repo touched provider/store: provider=%d memo=%d", len(author.Calls()), len(st.transitions)) + } + }) + } +} + +// TestForgeTransitionWorkflowStateOnGitHubIsInvalidArgument pins half the +// refinement/provider screen: workflow_state is a LINEAR refinement, so +// addressing GitHub with one is rejected at the ARM — never silently dropped, +// and never passed down for the provider to ignore. +func TestForgeTransitionWorkflowStateOnGitHubIsInvalidArgument(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + + fe := svc.ExecuteForgeCallAsAccountMust(t, transitionIssueCall("closed", "", "Done")).GetError() + if fe == nil || fe.GetCode() != "invalid_argument" { + t.Fatalf("workflow_state on GitHub error = %v, want invalid_argument", fe) + } + if !strings.Contains(fe.GetMessage(), "workflow_state") || !strings.Contains(fe.GetMessage(), "gh-author") { + t.Fatalf("message %q does not name the field and the provider", fe.GetMessage()) + } + if len(author.Calls()) != 0 || len(st.transitions) != 0 { + t.Fatalf("rejected refinement touched provider/store: provider=%d memo=%d", len(author.Calls()), len(st.transitions)) + } +} + +// TestForgeTransitionCloseReasonOnLinearIsInvalidArgument pins the other half: +// close_reason is a GitHub-ISSUE refinement, so addressing Linear with one is +// rejected at the arm. The mirrored case is what proves the screen is +// provider-directional rather than a single hard-coded rejection. +func TestForgeTransitionCloseReasonOnLinearIsInvalidArgument(t *testing.T) { + author := forge.NewFakeProvider("linear") + svc, st := newLinearForgeServiceForTest(t, author) + + fe := svc.ExecuteForgeCallAsAccountMust(t, transitionIssueCall("closed", "not_planned", "")).GetError() + if fe == nil || fe.GetCode() != "invalid_argument" { + t.Fatalf("close_reason on Linear error = %v, want invalid_argument", fe) + } + if !strings.Contains(fe.GetMessage(), "close_reason") || !strings.Contains(fe.GetMessage(), "linear") { + t.Fatalf("message %q does not name the field and the provider", fe.GetMessage()) + } + if len(author.Calls()) != 0 || len(st.transitions) != 0 { + t.Fatalf("rejected refinement touched provider/store: provider=%d memo=%d", len(author.Calls()), len(st.transitions)) + } +} + +// TestForgeTransitionLinearWorkflowStateReachesProvider is the screen's positive +// control: a refinement the ADDRESSED provider CAN express passes the screen and +// arrives in the dispatched input. Without this, a screen that rejected every +// refinement would pass the two rejection tests above. +func TestForgeTransitionLinearWorkflowStateReachesProvider(t *testing.T) { + author := forge.NewFakeProvider("linear") + svc, _ := newLinearForgeServiceForTest(t, author) + author.TransitionIssueResult = forge.Issue{Number: 7, State: "closed"} + + res := svc.ExecuteForgeCallAsAccountMust(t, transitionIssueCall("closed", "", "Done")) + if fe := res.GetError(); fe != nil { + t.Fatalf("Linear workflow_state was rejected: %v", fe) + } + calls := author.Calls() + if len(calls) != 1 { + t.Fatalf("author calls = %d, want 1", len(calls)) + } + in, ok := calls[0].Payload.(forge.TransitionState) + if !ok || in.WorkflowState != "Done" { + t.Fatalf("dispatched input = %+v, want WorkflowState=Done", calls[0].Payload) + } +} + +// TestForgeTransitionPRArmCarriesNoRefinements pins the PR-refinement screen +// structurally: the PR wire message exposes no refinement setter at all, so the +// only thing that can reach the provider is the portable state. Asserting the +// dispatched input is state-only is what catches a future arm that starts +// forwarding an issue refinement onto the PR path. +func TestForgeTransitionPRArmCarriesNoRefinements(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, _ := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + author.TransitionPRResult = forge.PullRequest{Number: 9, State: "open"} + + if fe := svc.ExecuteForgeCallAsAccountMust(t, transitionPRCall("open")).GetError(); fe != nil { + t.Fatalf("PR transition returned an in-band error: %v", fe) + } + in, ok := author.Calls()[0].Payload.(forge.TransitionState) + if !ok { + t.Fatalf("payload = %T, want forge.TransitionState", author.Calls()[0].Payload) + } + if in.CloseReason != "" || in.WorkflowState != "" { + t.Fatalf("PR arm forwarded a refinement: %+v", in) + } +} + +// TestForgeTransitionUnsupportedIsUnimplemented pins the flattening path both +// arms share: a provider with no PR model answers ErrUnsupported, which +// mapForgeError renders as an in-band unimplemented naming provider+op — and no +// memo lands, because the provider never succeeded. +func TestForgeTransitionUnsupportedIsUnimplemented(t *testing.T) { + author := forge.NewFakeProvider("linear") + svc, st := newLinearForgeServiceForTest(t, author) + author.SetError("TransitionPullRequestState", forge.ErrUnsupported) + + fe := svc.ExecuteForgeCallAsAccountMust(t, transitionPRCall("closed")).GetError() + if fe == nil || fe.GetCode() != "unimplemented" { + t.Fatalf("ErrUnsupported error = %v, want unimplemented", fe) + } + if !strings.Contains(fe.GetMessage(), "linear") || !strings.Contains(fe.GetMessage(), "transition_pull_request_state") { + t.Fatalf("unimplemented message %q does not name provider+op", fe.GetMessage()) + } + if len(st.transitions) != 0 { + t.Fatalf("memo writes = %d, want 0 (the provider failed)", len(st.transitions)) + } +} + +// --- tests: the actor memo ---------------------------------------------------- + +// TestForgeTransitionWritesActorMemoAfterProviderSuccess pins the §Actor +// attribution write half: on success the memo lands at the transitioned +// coordinate carrying the CALLING agent, the APPLIED portable state, and the +// chokepoint clock — and it lands STRICTLY AFTER the provider call, which the +// zero-memo-on-failure test below is the other half of. It also pins that a +// transition writes NO DL-055 ownership row: that row is a write-once authorship +// fact a transition must not touch. +func TestForgeTransitionWritesActorMemoAfterProviderSuccess(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + author.TransitionIssueResult = forge.Issue{Number: 7, State: "closed"} + + if fe := svc.ExecuteForgeCallAsAccountMust(t, transitionIssueCall("closed", "completed", "")).GetError(); fe != nil { + t.Fatalf("transition returned an in-band error: %v", fe) + } + if len(st.transitions) != 1 { + t.Fatalf("memo writes = %d, want 1", len(st.transitions)) + } + got := st.transitions[0] + want := recordedTransition{ + provider: store.ForgeProvider(compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB), + host: testHost, + repo: testRepo, + kind: store.ForgeArtifactKindIssue, + number: 7, + state: "closed", + agent: testAgentID, + at: nowStub(), + } + if got != want { + t.Fatalf("memo = %+v, want %+v", got, want) + } + if len(st.recorded) != 0 { + t.Fatalf("DL-055 rows = %d, want 0 (a transition mints no coordinate and must not touch the authorship row)", len(st.recorded)) + } +} + +// TestForgeTransitionPRMemoCarriesPortableStateAndPRKind pins the PR arm's memo: +// kind is pull_request, and the recorded state is the REQUESTED portable target, +// never the returned artifact's raw state — a merged PR reads back "merged", +// which is outside the portable domain the notify lane matches on. +func TestForgeTransitionPRMemoCarriesPortableStateAndPRKind(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + author.TransitionPRResult = forge.PullRequest{Number: 9, State: "merged"} + + if fe := svc.ExecuteForgeCallAsAccountMust(t, transitionPRCall("closed")).GetError(); fe != nil { + t.Fatalf("transition returned an in-band error: %v", fe) + } + if len(st.transitions) != 1 { + t.Fatalf("memo writes = %d, want 1", len(st.transitions)) + } + got := st.transitions[0] + if got.kind != store.ForgeArtifactKindPullRequest || got.number != 9 { + t.Fatalf("memo coordinate = kind %d number %d, want pull_request #9", got.kind, got.number) + } + if got.state != "closed" { + t.Fatalf("memo state = %q, want the requested portable %q (never the raw %q)", got.state, "closed", "merged") + } +} + +// TestForgeTransitionProviderFailureLeavesNoMemo pins the ordering the record +// arm holds too: the memo is written strictly AFTER provider success, so a +// rejected transition leaves NOTHING behind for the notify lane to attribute. +// Inverting the order would attribute a STATE event to an agent whose write the +// forge refused. +func TestForgeTransitionProviderFailureLeavesNoMemo(t *testing.T) { + for _, arm := range []struct { + name string + method string + call *compassv1internal.ForgeCallRequest + }{ + {"issue", "TransitionIssueState", transitionIssueCall("closed", "", "")}, + {"pull_request", "TransitionPullRequestState", transitionPRCall("closed")}, + } { + t.Run(arm.name, func(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + author.SetError(arm.method, &forge.StatusError{Status: 422, Message: "cannot reopen a merged pull request"}) + + fe := svc.ExecuteForgeCallAsAccountMust(t, arm.call).GetError() + if fe == nil || fe.GetCode() != "invalid_argument" { + t.Fatalf("422 error = %v, want invalid_argument", fe) + } + if len(author.Calls()) != 1 { + t.Fatalf("provider calls = %d, want 1 (the transition was attempted)", len(author.Calls())) + } + if len(st.transitions) != 0 { + t.Fatalf("memo writes = %d, want 0 (the provider rejected the transition)", len(st.transitions)) + } + }) + } +} + +// TestForgeTransitionMemoFailureAfterProviderSuccessIsInternal pins the +// memo-after-success fault path: the forge transition SUCCEEDED but the memo +// write failed, so the call returns an in-band internal error rather than a +// success the notify lane could never attribute. The forge-side state is already +// changed — the caller learning the attribution half failed is the honest answer. +// +// The MESSAGE is part of that honesty and is asserted here: it must name the +// coordinate and the applied state, so a caller cannot mistake a landed +// transition for a no-op and retry it, and a human reading the trace can see +// which half failed. +func TestForgeTransitionMemoFailureAfterProviderSuccessIsInternal(t *testing.T) { + author := forge.NewFakeProvider("gh-author") + svc, st := newForgeServiceForTest(t, author, forge.NewFakeProvider("gh-reviewer")) + author.TransitionIssueResult = forge.Issue{Number: 7, State: "closed"} + st.transErr = errors.New("memo: db unavailable") + + fe := svc.ExecuteForgeCallAsAccountMust(t, transitionIssueCall("closed", "", "")).GetError() + if fe == nil || fe.GetCode() != "internal" { + t.Fatalf("memo-after-success error = %v, want internal", fe) + } + for _, want := range []string{testRepo, "#7", "closed", "memo: db unavailable"} { + if !strings.Contains(fe.GetMessage(), want) { + t.Errorf("memo-after-success message %q does not name %q — the caller cannot tell the transition landed", fe.GetMessage(), want) + } + } + if len(author.Calls()) != 1 { + t.Fatalf("provider calls = %d, want 1 (the transition ran before the memo failed)", len(author.Calls())) + } + if len(st.transitions) != 0 { + t.Fatalf("memo writes = %d, want 0 (the write failed)", len(st.transitions)) + } +}