feat(linearagent): Linear Agent Session webhook types + API client (RIG-2717 T1/T2)

go/internal/store/accounts.go

@@ -141,6 +141,29 @@ func (s *Store) adminByHandle(ctx context.Context, handle string) (Account, erro
 // ErrConflict and fails startup — never silent adoption, mirroring
 // adminByHandle's posture.
 func (s *Store) EnsureSystemAccount(ctx context.Context) (Account, error) {
+	return s.ensureSystemSubtypeAccount(ctx, SystemAccountHandle, systemAccountDisplayName)
+}
+
+// EnsureLinearBridgeAccount ensures the reserved Linear bridge sender (@linear)
+// exists and returns it, idempotently — the author of Part 2 bridge posts. It
+// mints a SECOND system-subtype account beside @compass with the exact same
+// find-or-create shape (see ensureSystemSubtypeAccount): one accounts row plus a
+// system_accounts row on first boot, the existing row fetched on every later
+// boot. A pre-existing @linear row of the wrong shape is ErrConflict.
+func (s *Store) EnsureLinearBridgeAccount(ctx context.Context) (Account, error) {
+	return s.ensureSystemSubtypeAccount(ctx, LinearBridgeAccountHandle, linearBridgeDisplayName)
+}
+
+// ensureSystemSubtypeAccount is the shared find-or-create for a reserved
+// system-subtype account (handle + display name). Mirrors BootstrapAdmin's
+// unique-violation-means-fetch shape: on first boot it mints one accounts row
+// with a system_accounts subtype row — NOT a user or agent row; on every later
+// boot the insert hits the unique handle and the existing row is fetched and
+// returned. Its own insert is deliberately NOT routed through validateHandle:
+// this is the one path that mints a reserved handle. A pre-existing row of the
+// wrong shape (a user or agent row from a pre-guard database) is ErrConflict and
+// fails startup — never silent adoption, mirroring adminByHandle's posture.
+func (s *Store) ensureSystemSubtypeAccount(ctx context.Context, handle, displayName string) (Account, error) {
 	id := newID()
 	tx, err := s.pool.Begin(ctx)
 	if err != nil {
@@ -150,11 +173,11 @@ func (s *Store) EnsureSystemAccount(ctx context.Context) (Account, error) {
 
 	if _, err := tx.Exec(ctx,
 		"INSERT INTO accounts (id, handle, display_name) VALUES ($1, $2, $3)",
-		id, SystemAccountHandle, systemAccountDisplayName,
+		id, handle, displayName,
 	); err != nil {
 		if pgErrIs(err, pgUniqueViolation) {
 			// Already seeded (restart): fetch and return the existing system account.
-			return s.systemByHandle(ctx, SystemAccountHandle)
+			return s.systemByHandle(ctx, handle)
 		}
 		return Account{}, fmt.Errorf("store: insert account: %w", err)
 	}
@@ -169,8 +192,8 @@ func (s *Store) EnsureSystemAccount(ctx context.Context) (Account, error) {
 
 	return Account{
 		ID:          AccountID(id),
-		Handle:      SystemAccountHandle,
-		DisplayName: systemAccountDisplayName,
+		Handle:      handle,
+		DisplayName: displayName,
 		System:      &SystemAccount{},
 	}, nil
 }

go/internal/store/forge_authored.go

@@ -137,6 +137,37 @@ func (s *Store) AuthoredArtifactByRequestID(ctx context.Context, agent AccountID
 	return a, true, nil
 }
 
+// AuthoredArtifactByCoordinate reads the ownership row at a forge coordinate —
+// the by-coordinate lookup T4 uses to resolve a delegated issue's recorded
+// authoring agent (design compass-linear-agent-responder §Part 2). The forge
+// coordinate (provider, host, repo, kind, number) IS the
+// forge_authored_artifacts PK, so this is a trivial PK lookup: an unknown
+// coordinate is ErrNotFound; zero/empty coordinate fields (or a zero kind) are
+// ErrInvalidArgument, mirroring the validation RecordAuthoredArtifact applies.
+func (s *Store) AuthoredArtifactByCoordinate(ctx context.Context, provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64) (AuthoredArtifact, error) {
+	if err := validCoordinate(provider, host, repo); err != nil {
+		return AuthoredArtifact{}, err
+	}
+	if kind == ForgeArtifactKindUnspecified {
+		return AuthoredArtifact{}, fmt.Errorf("%w: artifact kind is required", ErrInvalidArgument)
+	}
+	row := s.pool.QueryRow(ctx,
+		`SELECT forge_provider, forge_host, repo, kind, number,
+		        agent_account_id, owner_user_id, session_id, client_request_id, created_at_unix_ms
+		   FROM forge_authored_artifacts
+		  WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND number = $5`,
+		int32(provider), host, repo, int32(kind), 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.
+	)
+	a, err := scanAuthoredArtifact(row)
+	if err != nil {
+		if noRows(err) {
+			return AuthoredArtifact{}, fmt.Errorf("%w: authored artifact at coordinate %d/%s/%s kind %d number %d", ErrNotFound, provider, host, repo, kind, number)
+		}
+		return AuthoredArtifact{}, fmt.Errorf("store: read authored artifact by coordinate: %w", err)
+	}
+	return a, nil
+}
+
 // ListAuthoredArtifactsByAgent reads every artifact the agent authored, ordered
 // deterministically by created_at then coordinate. No rows is a nil slice, not
 // an error. Zero agent -> ErrInvalidArgument.

go/internal/store/handle.go

@@ -15,14 +15,16 @@ import (
 var handleRE = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*$`)
 
 // reservedHandles are handles no account may register. `compass` is the system
-// handle; `everyone`/`agents`/`users` are the server's reserved broadcast
-// mentions — an account registered as one of those would shadow a live
-// broadcast semantic. The reserved-mention names duplicate delivery's
-// reservedMentions (consumer.go:326, the source of truth) rather than importing
-// the delivery package, to avoid a store->delivery dependency; keep them in
-// sync.
+// handle and `linear` the Linear bridge system handle (both seeded as
+// system-subtype accounts, never registrable by a user/agent);
+// `everyone`/`agents`/`users` are the server's reserved broadcast mentions — an
+// account registered as one of those would shadow a live broadcast semantic.
+// The reserved-mention names duplicate delivery's reservedMentions
+// (consumer.go:326, the source of truth) rather than importing the delivery
+// package, to avoid a store->delivery dependency; keep them in sync.
 var reservedHandles = map[string]bool{
 	"compass":  true,
+	"linear":   true,
 	"everyone": true,
 	"agents":   true,
 	"users":    true,

go/internal/store/linear_sessions.go

@@ -0,0 +1,98 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// The Linear Agent Session association (compass-linear-agent-responder
+// design.md §Part 2 / §T3): the durable link between a Linear AgentSession and
+// the Compass conversation the responder routed it to. Written on a `created`
+// event (UpsertLinearAgentSession, idempotent on the session-id PK) and read on
+// a `prompted` event (LinearAgentSession) to route the follow-up to the same
+// Manager/topic. No dedup column — message-level dedup is the comms rail's
+// client_request_id (§Part 1); the association insert is idempotent on its own.
+
+// LinearAgentSessionRow is one association row: the Linear session id, the Compass
+// Manager the delegated issue routed to, that Manager's home channel, the comms
+// topic the conversation landed in, and the issue it was delegated on
+// (provenance; "" when none). CreatedAt is the server-assigned birth time.
+type LinearAgentSessionRow struct {
+	LinearSessionID  string
+	ManagerAccountID AccountID
+	ChannelID        ChannelID
+	TopicID          string
+	LinearIssueID    string // provenance; "" = no issue recorded (stored as SQL NULL)
+	CreatedAt        time.Time
+}
+
+// UpsertLinearAgentSession idempotently records the association at row's
+// linear_session_id: INSERT … ON CONFLICT (linear_session_id) DO NOTHING. It
+// returns created=true when this call inserted the row and created=false on a
+// replay (the session was already associated) — the caller uses that to skip
+// the one-time `created`-side work (routing, ack thought, deep link) on a
+// redelivered `created` event. An empty linear_session_id is a caller bug
+// (ErrInvalidArgument). LinearIssueID "" is stored as SQL NULL.
+func (s *Store) UpsertLinearAgentSession(ctx context.Context, row LinearAgentSessionRow) (created bool, err error) {
+	if row.LinearSessionID == "" {
+		return false, fmt.Errorf("%w: linear session id is required", ErrInvalidArgument)
+	}
+	tag, err := s.pool.Exec(ctx,
+		`INSERT INTO linear_agent_sessions
+		     (linear_session_id, manager_account_id, channel_id, topic_id, linear_issue_id)
+		 VALUES ($1, $2, $3, $4, $5)
+		 ON CONFLICT (linear_session_id) DO NOTHING`,
+		row.LinearSessionID, string(row.ManagerAccountID), string(row.ChannelID),
+		row.TopicID, nullIfEmpty(row.LinearIssueID),
+	)
+	if err != nil {
+		return false, fmt.Errorf("store: upsert linear agent session: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// LinearAgentSession reads the association for linearSessionID — the `prompted`
+// lookup that routes a follow-up to the recorded Manager/topic. An unknown
+// session id is ErrNotFound; an empty id is ErrInvalidArgument.
+func (s *Store) LinearAgentSession(ctx context.Context, linearSessionID string) (LinearAgentSessionRow, error) {
+	if linearSessionID == "" {
+		return LinearAgentSessionRow{}, fmt.Errorf("%w: linear session id is required", ErrInvalidArgument)
+	}
+	row := s.pool.QueryRow(ctx,
+		`SELECT linear_session_id, manager_account_id, channel_id, topic_id, linear_issue_id, created_at
+		   FROM linear_agent_sessions
+		  WHERE linear_session_id = $1`,
+		linearSessionID,
+	)
+	r, err := scanLinearAgentSession(row)
+	if err != nil {
+		if noRows(err) {
+			return LinearAgentSessionRow{}, fmt.Errorf("%w: linear agent session %q", ErrNotFound, linearSessionID)
+		}
+		return LinearAgentSessionRow{}, fmt.Errorf("store: read linear agent session: %w", err)
+	}
+	return r, nil
+}
+
+// scanLinearAgentSession scans one row into a LinearAgentSessionRow, mapping the
+// nullable linear_issue_id column to "" (no issue) via a pgx-native scan.
+func scanLinearAgentSession(row pgx.Row) (LinearAgentSessionRow, error) {
+	var (
+		r       LinearAgentSessionRow
+		manager string
+		channel string
+		issueID *string
+	)
+	if err := row.Scan(&r.LinearSessionID, &manager, &channel, &r.TopicID, &issueID, &r.CreatedAt); err != nil {
+		return LinearAgentSessionRow{}, err
+	}
+	r.ManagerAccountID = AccountID(manager)
+	r.ChannelID = ChannelID(channel)
+	if issueID != nil {
+		r.LinearIssueID = *issueID
+	}
+	return r, nil
+}

go/internal/store/linear_sessions_pgtest_test.go

@@ -0,0 +1,147 @@
+//go:build pgtest
+
+package store
+
+// Linear Agent Session association + by-coordinate ownership read store
+// contracts (compass-linear-agent-responder §T3): the linear_agent_sessions
+// table shape, the idempotent upsert (created=true first, false on replay), the
+// PK lookup (hit + ErrNotFound miss), and AuthoredArtifactByCoordinate
+// (seeded-row hit + ErrNotFound miss). context.Background is the test root
+// (the pgtest-suite convention, sibling forge_authored_pgtest_test.go).
+
+import (
+	"context"
+	"testing"
+)
+
+// ── Upsert + lookup round-trip; replay returns created=false ──────────────────
+
+func TestUpsertLinearAgentSession(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+
+	row := LinearAgentSessionRow{
+		LinearSessionID:  "sess-abc",
+		ManagerAccountID: "mgr-1",
+		ChannelID:        "chan-1",
+		TopicID:          "topic-1",
+		LinearIssueID:    "issue-1",
+	}
+	created, err := s.UpsertLinearAgentSession(ctx, row)
+	if err != nil {
+		t.Fatalf("first upsert: %v", err)
+	}
+	if !created {
+		t.Fatalf("first upsert created = false, want true")
+	}
+
+	// Replay the SAME session id (ON CONFLICT DO NOTHING): no error, created=false.
+	replay := row
+	replay.ManagerAccountID = "mgr-CHANGED" // DO NOTHING must not rewrite it
+	created, err = s.UpsertLinearAgentSession(ctx, replay)
+	if err != nil {
+		t.Fatalf("replay upsert: %v", err)
+	}
+	if created {
+		t.Fatalf("replay upsert created = true, want false")
+	}
+
+	// The original row survives the replay unchanged.
+	got, err := s.LinearAgentSession(ctx, "sess-abc")
+	if err != nil {
+		t.Fatalf("lookup: %v", err)
+	}
+	if got.LinearSessionID != row.LinearSessionID ||
+		got.ManagerAccountID != row.ManagerAccountID ||
+		got.ChannelID != row.ChannelID ||
+		got.TopicID != row.TopicID ||
+		got.LinearIssueID != row.LinearIssueID {
+		t.Fatalf("read-back = %+v, want %+v (replay must not clobber)", got, row)
+	}
+	if got.CreatedAt.IsZero() {
+		t.Fatalf("created_at is zero, want the DEFAULT now() birth time")
+	}
+}
+
+// ── Lookup miss → ErrNotFound; empty id → ErrInvalidArgument ───────────────────
+
+func TestLinearAgentSession(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+
+	_, err := s.LinearAgentSession(ctx, "no-such-session")
+	sentinelIs(t, err, ErrNotFound, "lookup unknown session")
+
+	_, err = s.LinearAgentSession(ctx, "")
+	sentinelIs(t, err, ErrInvalidArgument, "lookup empty session id")
+
+	// Empty LinearIssueID stores as SQL NULL and reads back as "".
+	if _, err := s.UpsertLinearAgentSession(ctx, LinearAgentSessionRow{
+		LinearSessionID:  "sess-noissue",
+		ManagerAccountID: "mgr-1",
+		ChannelID:        "chan-1",
+		TopicID:          "topic-1",
+	}); err != nil {
+		t.Fatalf("upsert no-issue: %v", err)
+	}
+	got, err := s.LinearAgentSession(ctx, "sess-noissue")
+	if err != nil {
+		t.Fatalf("lookup no-issue: %v", err)
+	}
+	if got.LinearIssueID != "" {
+		t.Fatalf("linear_issue_id = %q, want \"\" (NULL → empty)", got.LinearIssueID)
+	}
+
+	// Empty session id on the write path is a caller bug too.
+	if _, err := s.UpsertLinearAgentSession(ctx, LinearAgentSessionRow{}); err == nil {
+		t.Fatalf("upsert empty session id: want ErrInvalidArgument, got nil")
+	} else {
+		sentinelIs(t, err, ErrInvalidArgument, "upsert empty session id")
+	}
+}
+
+// ── AuthoredArtifactByCoordinate: seeded-row hit + ErrNotFound miss ────────────
+
+func TestAuthoredArtifactByCoordinate(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	agent, owner := seedAgent(t, s, "coord")
+
+	want := AuthoredArtifact{
+		Provider:        ForgeProviderGitHub,
+		Host:            "github.com",
+		Repo:            "a/b",
+		Kind:            ForgeArtifactKindIssue,
+		Number:          7,
+		AgentAccountID:  agent,
+		OwnerUserID:     owner,
+		SessionID:       "sess-1",
+		ClientRequestID: "req-1",
+		CreatedAtUnixMS: 1000,
+	}
+	if err := s.RecordAuthoredArtifact(ctx, want); err != nil {
+		t.Fatalf("record: %v", err)
+	}
+
+	got, err := s.AuthoredArtifactByCoordinate(ctx, want.Provider, want.Host, want.Repo, want.Kind, want.Number)
+	if err != nil {
+		t.Fatalf("by-coordinate hit: %v", err)
+	}
+	if got != want {
+		t.Fatalf("by-coordinate read = %+v, want %+v", got, want)
+	}
+
+	// Unknown coordinate (same repo, different number) → ErrNotFound.
+	_, err = s.AuthoredArtifactByCoordinate(ctx, want.Provider, want.Host, want.Repo, want.Kind, 999)
+	sentinelIs(t, err, ErrNotFound, "by-coordinate unknown number")
+
+	// A wrong kind at the same number is a distinct coordinate → miss.
+	_, err = s.AuthoredArtifactByCoordinate(ctx, want.Provider, want.Host, want.Repo, ForgeArtifactKindPullRequest, want.Number)
+	sentinelIs(t, err, ErrNotFound, "by-coordinate wrong kind")
+
+	// Invalid coordinate fields → ErrInvalidArgument.
+	_, err = s.AuthoredArtifactByCoordinate(ctx, ForgeProviderUnspecified, want.Host, want.Repo, want.Kind, want.Number)
+	sentinelIs(t, err, ErrInvalidArgument, "by-coordinate zero provider")
+	_, err = s.AuthoredArtifactByCoordinate(ctx, want.Provider, want.Host, want.Repo, ForgeArtifactKindUnspecified, want.Number)
+	sentinelIs(t, err, ErrInvalidArgument, "by-coordinate zero kind")
+}

go/internal/store/migrations/0001_init.sql

@@ -729,3 +729,31 @@ CREATE UNIQUE INDEX forge_authored_artifacts_request_memo_idx
 -- By-agent scan (ListAuthoredArtifactsByAgent): every artifact one agent authored.
 CREATE INDEX forge_authored_artifacts_agent_idx
     ON forge_authored_artifacts (agent_account_id);
+
+-- 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
+-- Manager's home channel, the comms topic the delegated conversation landed in,
+-- and the issue it was delegated on (provenance). Read on a `prompted` event to
+-- route the follow-up to the same topic (LinearAgentSession); written on
+-- `created` (UpsertLinearAgentSession, ON CONFLICT DO NOTHING).
+--
+-- NO dedup column: message-level dedup is the comms rail's client_request_id
+-- (keyed on the Linear-Delivery UUID, §Part 1), not this table's concern. The
+-- association insert is itself idempotent on the linear_session_id PK, so a
+-- replayed `created` re-lands on the key rather than forking a second row.
+--
+-- text ids are server/forge-assigned; created_at is a TIMESTAMPTZ DEFAULT now()
+-- birth marker. No FKs: manager_account_id, channel_id and topic_id name live
+-- Compass rows, but the association is written from the webhook path against ids
+-- the responder just resolved, and a Manager/channel/topic teardown must not be
+-- blocked by a stale Linear association — so these are unconstrained id columns,
+-- matching the schema in the record (§Part 2).
+CREATE TABLE linear_agent_sessions (
+    linear_session_id  TEXT PRIMARY KEY,               -- Linear AgentSession.id
+    manager_account_id TEXT NOT NULL,                  -- the resolved Compass Manager
+    channel_id         TEXT NOT NULL,                  -- the Manager's home channel
+    topic_id           TEXT NOT NULL,                  -- comms topic of the conversation
+    linear_issue_id    TEXT,                           -- provenance (issue delegated on); NULL if none
+    created_at         TIMESTAMPTZ NOT NULL DEFAULT now()
+);