diff --git a/go/internal/store/forge_subscriptions.go b/go/internal/store/forge_subscriptions.go
new file mode 100644
index 00000000..8387b8d4
--- /dev/null
+++ b/go/internal/store/forge_subscriptions.go
@@ -0,0 +1,166 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// The DL-053 agent-notification subscription writer (RIG-2732 Piece 1, design
+// docs/designs/product/compass-notification-delivery/design.md): the Server-side
+// Postgres row that records an agent's standing interest in one forge artifact.
+// agent_forge_subscriptions is the per-subscriber DELIVERY-cursor table
+// (delivered_revision/delivered_at); forge_artifact_cursors is the shared
+// per-artifact FETCH cursor the poll driver writes. The GC invariant here is the
+// only place this slice touches forge_artifact_cursors: when the LAST
+// subscription for a coordinate is deleted, its cursor row is collected in the
+// same transaction (DL-053). The poll driver owns the cursor WRITER (Piece 2);
+// this file never inserts a cursor row.
+
+// AgentForgeSubscription is one row of agent_forge_subscriptions: an agent's
+// standing interest in one forge artifact coordinate, plus that subscriber's
+// per-artifact DELIVERY cursor (DeliveredRevision/DeliveredAt — the last
+// revision this agent was notified of, distinct from the shared FETCH cursor on
+// forge_artifact_cursors). Mirrors ForgeRepoSubscription's field style
+// (forge_cursors.go).
+type AgentForgeSubscription struct {
+	ID             string
+	AgentAccountID AccountID
+	Provider       ForgeProvider // GITHUB(1)/GITLAB(2)/FORGEJO(3)/LINEAR(4); never 0
+	Host           string
+	Repo           string
+	Kind           ForgeArtifactKind // issue(1)/pull_request(2); never 0
+	Number         uint64
+
+	DeliveredRevision string
+	DeliveredAt       *time.Time
+	CreatedAt         time.Time
+}
+
+// validSubscriptionCoordinate rejects the zero/empty coordinate fields
+// EnsureAgentForgeSubscription guards on before any DB round trip: the
+// provider/host/repo triple (via validCoordinate), a zero kind (never
+// UNSPECIFIED(0), the CHECK's job in Go space), and a zero artifact number. A
+// caller bug is ErrInvalidArgument.
+func validSubscriptionCoordinate(provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64) error {
+	if err := validCoordinate(provider, host, repo); err != nil {
+		return err
+	}
+	if kind != ForgeArtifactKindIssue && kind != ForgeArtifactKindPullRequest {
+		return fmt.Errorf("%w: artifact kind must be issue or pull_request", ErrInvalidArgument)
+	}
+	if number == 0 {
+		return fmt.Errorf("%w: artifact number is required", ErrInvalidArgument)
+	}
+	return nil
+}
+
+// EnsureAgentForgeSubscription idempotently inserts the agent's subscription to
+// one artifact coordinate, keyed by the UNIQUE (agent_account_id, provider,
+// host, repo, kind, number). A repeat subscribe by the same agent to the same
+// artifact returns the EXISTING subscription id and creates no duplicate row —
+// the DO UPDATE is a no-op touch (re-setting agent_account_id to itself) that
+// makes RETURNING fire on the conflict path so a repeat returns the stored id,
+// not a fresh one. A new coordinate mints a fresh id via newID(). Zero/empty
+// coordinate fields / a zero kind / a zero number -> ErrInvalidArgument; an
+// unknown agent (the FK RESTRICT) -> ErrInvalidArgument.
+func (s *Store) EnsureAgentForgeSubscription(ctx context.Context, sub AgentForgeSubscription) (string, error) {
+	if err := validSubscriptionCoordinate(sub.Provider, sub.Host, sub.Repo, sub.Kind, sub.Number); err != nil {
+		return "", err
+	}
+	if sub.AgentAccountID == "" {
+		return "", fmt.Errorf("%w: agent account id is required", ErrInvalidArgument)
+	}
+	var id string
+	if err := s.pool.QueryRow(ctx,
+		`INSERT INTO agent_forge_subscriptions
+		     (id, agent_account_id, forge_provider, forge_host, repo, kind, number)
+		 VALUES ($1, $2, $3, $4, $5, $6, $7)
+		 ON CONFLICT (agent_account_id, forge_provider, forge_host, repo, kind, number) DO UPDATE
+		    SET agent_account_id = EXCLUDED.agent_account_id
+		 RETURNING id`,
+		newID(), string(sub.AgentAccountID), int32(sub.Provider), sub.Host, sub.Repo,
+		int32(sub.Kind), int64(sub.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.
+	).Scan(&id); err != nil {
+		if pgErrIs(err, pgForeignKeyViolation) {
+			return "", fmt.Errorf("%w: unknown agent %q", ErrInvalidArgument, sub.AgentAccountID)
+		}
+		return "", fmt.Errorf("store: ensure agent forge subscription: %w", err)
+	}
+	return id, nil
+}
+
+// DeleteAgentForgeSubscription removes the agent's subscription by id, scoped to
+// the calling agent (an agent cannot delete another agent's subscription — the
+// WHERE clause matches on both id AND agent). Zero rows deleted (an unknown id,
+// or an id owned by a different agent, indistinguishable by design) -> ErrNotFound.
+//
+// The delete runs the DL-053 garbage-collection invariant in ONE transaction:
+// after removing the subscription row, its coordinate's forge_artifact_cursors
+// row is collected IFF that was the last subscription for the coordinate — the
+// NOT EXISTS guard leaves the cursor in place if any other agent still
+// subscribes to the same artifact. The coordinate is taken from the deleted
+// row's RETURNING, so the GC targets exactly the artifact whose last subscriber
+// just left.
+func (s *Store) DeleteAgentForgeSubscription(ctx context.Context, agent AccountID, subscriptionID string) error {
+	if agent == "" {
+		return fmt.Errorf("%w: agent account id is required", ErrInvalidArgument)
+	}
+	if subscriptionID == "" {
+		return fmt.Errorf("%w: subscription id is required", ErrInvalidArgument)
+	}
+	return s.WithTx(ctx, func(tx pgx.Tx) error {
+		var (
+			provider int32
+			host     string
+			repo     string
+			kind     int32
+			number   int64
+		)
+		if err := tx.QueryRow(ctx,
+			`DELETE FROM agent_forge_subscriptions
+			  WHERE id = $1 AND agent_account_id = $2
+			 RETURNING forge_provider, forge_host, repo, kind, number`,
+			subscriptionID, string(agent),
+		).Scan(&provider, &host, &repo, &kind, &number); err != nil {
+			if noRows(err) {
+				return fmt.Errorf("%w: subscription %q", ErrNotFound, subscriptionID)
+			}
+			return fmt.Errorf("store: delete agent forge subscription: %w", err)
+		}
+		if _, err := tx.Exec(ctx,
+			`DELETE FROM forge_artifact_cursors
+			  WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND number = $5
+			    AND NOT EXISTS (
+			        SELECT 1 FROM agent_forge_subscriptions
+			         WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND number = $5
+			    )`,
+			provider, host, repo, kind, number,
+		); err != nil {
+			return fmt.Errorf("store: garbage-collect forge artifact cursor: %w", err)
+		}
+		return nil
+	})
+}
+
+// AgentForgeSubscriptionsForArtifact counts the subscriptions currently held on
+// one artifact coordinate — the minimal reader the GC test needs to assert the
+// last-subscription-deletes-cursor invariant. It is NOT the poll driver's
+// subscriber-enumeration reader (Piece 2); it returns only the row count.
+// Zero/empty coordinate fields / a zero kind / a zero number -> ErrInvalidArgument.
+func (s *Store) AgentForgeSubscriptionsForArtifact(ctx context.Context, provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64) (int, error) {
+	if err := validSubscriptionCoordinate(provider, host, repo, kind, number); err != nil {
+		return 0, err
+	}
+	var n int
+	if err := s.pool.QueryRow(ctx,
+		`SELECT count(*) FROM agent_forge_subscriptions
+		  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 written to a BIGINT, always within the int64 domain.
+	).Scan(&n); err != nil {
+		return 0, fmt.Errorf("store: count agent forge subscriptions for artifact: %w", err)
+	}
+	return n, nil
+}
diff --git a/go/internal/store/forge_subscriptions_pgtest_test.go b/go/internal/store/forge_subscriptions_pgtest_test.go
new file mode 100644
index 00000000..94c11dc3
--- /dev/null
+++ b/go/internal/store/forge_subscriptions_pgtest_test.go
@@ -0,0 +1,257 @@
+//go:build pgtest
+
+package store
+
+// DL-053 agent-forge-subscription store contracts (RIG-2732 Piece 1): the
+// idempotent per-agent subscribe on the UNIQUE (agent, coordinate), the
+// id+agent-scoped delete, and the load-bearing GC invariant — the
+// forge_artifact_cursors row is collected exactly when the LAST subscription for
+// its coordinate is deleted, and survives while any other agent still
+// subscribes. context.Background is the test root (the pgtest-suite convention,
+// sibling forge_authored_pgtest_test.go).
+
+import (
+	"context"
+	"testing"
+)
+
+// artifactCursorExists reports whether a forge_artifact_cursors row is present
+// at the coordinate — the GC assertion surface (no public reader for the
+// poll-driver's cursor table in this slice).
+func artifactCursorExists(t *testing.T, s *Store, provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64) bool {
+	t.Helper()
+	var n int
+	if err := s.pool.QueryRow(context.Background(),
+		`SELECT count(*) FROM forge_artifact_cursors
+		  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),
+	).Scan(&n); err != nil {
+		t.Fatalf("count artifact cursor rows: %v", err)
+	}
+	return n > 0
+}
+
+// seedArtifactCursor inserts a bare forge_artifact_cursors row at the coordinate
+// (the poll driver's WRITER is a later slice, so the GC test seeds directly).
+func seedArtifactCursor(t *testing.T, s *Store, provider ForgeProvider, host, repo string, kind ForgeArtifactKind, number uint64) {
+	t.Helper()
+	if _, err := s.pool.Exec(context.Background(),
+		`INSERT INTO forge_artifact_cursors (forge_provider, forge_host, repo, kind, number)
+		 VALUES ($1, $2, $3, $4, $5)`,
+		int32(provider), host, repo, int32(kind), int64(number),
+	); err != nil {
+		t.Fatalf("seed artifact cursor: %v", err)
+	}
+}
+
+// ── Test 1: idempotent subscribe on the UNIQUE (agent, coordinate) ────────────
+
+func TestAgentForgeSubscriptionIdempotent(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	agent, _ := seedAgent(t, s, "sub1")
+
+	base := AgentForgeSubscription{
+		AgentAccountID: agent,
+		Provider:       ForgeProviderGitHub,
+		Host:           "github.com",
+		Repo:           "a/b",
+		Kind:           ForgeArtifactKindIssue,
+		Number:         42,
+	}
+	first, err := s.EnsureAgentForgeSubscription(ctx, base)
+	if err != nil {
+		t.Fatalf("first ensure: %v", err)
+	}
+	if first == "" {
+		t.Fatalf("first ensure returned empty id")
+	}
+
+	second, err := s.EnsureAgentForgeSubscription(ctx, base)
+	if err != nil {
+		t.Fatalf("second ensure: %v", err)
+	}
+	if second != first {
+		t.Fatalf("repeat subscribe id = %q, want %q (idempotent)", second, first)
+	}
+
+	n, err := s.AgentForgeSubscriptionsForArtifact(ctx, base.Provider, base.Host, base.Repo, base.Kind, base.Number)
+	if err != nil {
+		t.Fatalf("count: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("row count = %d, want 1 (no duplicate)", n)
+	}
+}
+
+// ── Test 2: two agents on the same coordinate → two distinct rows ─────────────
+
+func TestAgentForgeSubscriptionDistinctPerAgent(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	agentA, _ := seedAgent(t, s, "sub2a")
+	agentB, _ := seedAgent(t, s, "sub2b")
+
+	coord := AgentForgeSubscription{
+		Provider: ForgeProviderGitHub, Host: "github.com", Repo: "a/b",
+		Kind: ForgeArtifactKindIssue, Number: 7,
+	}
+	a := coord
+	a.AgentAccountID = agentA
+	b := coord
+	b.AgentAccountID = agentB
+
+	idA, err := s.EnsureAgentForgeSubscription(ctx, a)
+	if err != nil {
+		t.Fatalf("ensure A: %v", err)
+	}
+	idB, err := s.EnsureAgentForgeSubscription(ctx, b)
+	if err != nil {
+		t.Fatalf("ensure B: %v", err)
+	}
+	if idA == idB {
+		t.Fatalf("two agents share id %q, want distinct", idA)
+	}
+	n, err := s.AgentForgeSubscriptionsForArtifact(ctx, coord.Provider, coord.Host, coord.Repo, coord.Kind, coord.Number)
+	if err != nil {
+		t.Fatalf("count: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("row count = %d, want 2", n)
+	}
+}
+
+// ── Test 3: delete is id+agent-scoped ─────────────────────────────────────────
+
+func TestAgentForgeSubscriptionDeleteScoping(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	agentA, _ := seedAgent(t, s, "sub3a")
+	agentB, _ := seedAgent(t, s, "sub3b")
+
+	id, err := s.EnsureAgentForgeSubscription(ctx, AgentForgeSubscription{
+		AgentAccountID: agentA, Provider: ForgeProviderGitHub, Host: "github.com",
+		Repo: "a/b", Kind: ForgeArtifactKindIssue, Number: 1,
+	})
+	if err != nil {
+		t.Fatalf("ensure: %v", err)
+	}
+
+	// Deleting an unknown id -> ErrNotFound.
+	sentinelIs(t, s.DeleteAgentForgeSubscription(ctx, agentA, "no-such-id"), ErrNotFound, "delete unknown id")
+	// Deleting another agent's id -> ErrNotFound (scoping), row survives.
+	sentinelIs(t, s.DeleteAgentForgeSubscription(ctx, agentB, id), ErrNotFound, "delete foreign id")
+	if n, _ := s.AgentForgeSubscriptionsForArtifact(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 1); n != 1 {
+		t.Fatalf("row count after foreign delete = %d, want 1 (untouched)", n)
+	}
+	// Owner deletes -> gone.
+	if err := s.DeleteAgentForgeSubscription(ctx, agentA, id); err != nil {
+		t.Fatalf("owner delete: %v", err)
+	}
+	if n, _ := s.AgentForgeSubscriptionsForArtifact(ctx, ForgeProviderGitHub, "github.com", "a/b", ForgeArtifactKindIssue, 1); n != 0 {
+		t.Fatalf("row count after owner delete = %d, want 0", n)
+	}
+}
+
+// ── Test 4: DL-053 GC — cursor collected on LAST unsubscribe only ─────────────
+
+func TestAgentForgeSubscriptionCursorGC(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	agentA, _ := seedAgent(t, s, "gc-a")
+	agentB, _ := seedAgent(t, s, "gc-b")
+
+	const (
+		host   = "github.com"
+		repo   = "a/b"
+		number = uint64(99)
+	)
+	provider := ForgeProviderGitHub
+	kind := ForgeArtifactKindPullRequest
+
+	seedArtifactCursor(t, s, provider, host, repo, kind, number)
+
+	idA, err := s.EnsureAgentForgeSubscription(ctx, AgentForgeSubscription{
+		AgentAccountID: agentA, Provider: provider, Host: host, Repo: repo, Kind: kind, Number: number,
+	})
+	if err != nil {
+		t.Fatalf("ensure A: %v", err)
+	}
+	idB, err := s.EnsureAgentForgeSubscription(ctx, AgentForgeSubscription{
+		AgentAccountID: agentB, Provider: provider, Host: host, Repo: repo, Kind: kind, Number: number,
+	})
+	if err != nil {
+		t.Fatalf("ensure B: %v", err)
+	}
+
+	// Delete A: B still subscribes -> cursor STAYS.
+	if err := s.DeleteAgentForgeSubscription(ctx, agentA, idA); err != nil {
+		t.Fatalf("delete A: %v", err)
+	}
+	if !artifactCursorExists(t, s, provider, host, repo, kind, number) {
+		t.Fatalf("cursor collected after A left, but B still subscribes")
+	}
+
+	// Delete B: last subscription gone -> cursor COLLECTED.
+	if err := s.DeleteAgentForgeSubscription(ctx, agentB, idB); err != nil {
+		t.Fatalf("delete B: %v", err)
+	}
+	if artifactCursorExists(t, s, provider, host, repo, kind, number) {
+		t.Fatalf("cursor survived after last subscription deleted (DL-053 GC failed)")
+	}
+}
+
+// ── Test 5: coordinate validation ─────────────────────────────────────────────
+
+func TestAgentForgeSubscriptionValidation(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	agent, _ := seedAgent(t, s, "sub5")
+
+	good := AgentForgeSubscription{
+		AgentAccountID: agent, Provider: ForgeProviderGitHub, Host: "github.com",
+		Repo: "a/b", Kind: ForgeArtifactKindIssue, Number: 1,
+	}
+	cases := []struct {
+		name  string
+		mutfn func(*AgentForgeSubscription)
+	}{
+		{"zero provider", func(a *AgentForgeSubscription) { a.Provider = ForgeProviderUnspecified }},
+		{"empty host", func(a *AgentForgeSubscription) { a.Host = "" }},
+		{"empty repo", func(a *AgentForgeSubscription) { a.Repo = "" }},
+		{"zero kind", func(a *AgentForgeSubscription) { a.Kind = ForgeArtifactKindUnspecified }},
+		{"zero number", func(a *AgentForgeSubscription) { a.Number = 0 }},
+		{"empty agent", func(a *AgentForgeSubscription) { a.AgentAccountID = "" }},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			bad := good
+			tc.mutfn(&bad)
+			_, err := s.EnsureAgentForgeSubscription(ctx, bad)
+			sentinelIs(t, err, ErrInvalidArgument, tc.name)
+		})
+	}
+}
+
+// TestAgentForgeSubscriptionUnknownAgentIsInvalidArgument covers the FK-violation
+// branch of EnsureAgentForgeSubscription: a well-formed coordinate naming an
+// agent that does not exist trips the agent_account_id FK RESTRICT, which the
+// writer must classify as ErrInvalidArgument (rendered in-band as invalid_argument
+// via storeForgeError) — NOT a raw pg error / CodeInternal teardown. The sibling
+// TestAgentForgeSubscriptionFKRestrict asserts the DB constraint fires on a raw
+// INSERT; this asserts the METHOD's error-classification branch, which that test
+// bypasses.
+func TestAgentForgeSubscriptionUnknownAgentIsInvalidArgument(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+
+	_, err := s.EnsureAgentForgeSubscription(ctx, AgentForgeSubscription{
+		AgentAccountID: AccountID("no-such-agent"),
+		Provider:       ForgeProviderGitHub,
+		Host:           "github.com",
+		Repo:           "a/b",
+		Kind:           ForgeArtifactKindIssue,
+		Number:         1,
+	})
+	sentinelIs(t, err, ErrInvalidArgument, "unknown agent (FK violation)")
+}
diff --git a/go/server/forge.go b/go/server/forge.go
index 080cf9e9..6db2bc8a 100644
--- a/go/server/forge.go
+++ b/go/server/forge.go
@@ -141,6 +141,8 @@ 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
 }
 
 // forgeService is the ForgeCaller implementation and the DL-050 write
@@ -203,13 +205,9 @@ func (s *forgeService) ExecuteForgeCallAsAccount(
 	case *compassv1internal.ForgeCallRequest_GetPullRequest:
 		return s.getPullRequest(ctx, call, c.GetPullRequest), nil
 	case *compassv1internal.ForgeCallRequest_Subscribe:
-		// Row writes on agent_forge_subscriptions are the poll-driver lane's
-		// surface (DL-053/A8); no store writer for that table exists in this
-		// slice, so the arm is unimplemented rather than inventing a store
-		// method (see summary: subscription-writer dependency).
-		return forgeErrorResult(forgeErr(connect.CodeUnimplemented, "forge: subscribe is not wired (no agent_forge_subscriptions store writer yet)")), nil
+		return s.subscribeForge(ctx, caller, call, c.Subscribe), nil
 	case *compassv1internal.ForgeCallRequest_Unsubscribe:
-		return forgeErrorResult(forgeErr(connect.CodeUnimplemented, "forge: unsubscribe is not wired (no agent_forge_subscriptions store writer yet)")), nil
+		return s.unsubscribeForge(ctx, caller, c.Unsubscribe), nil
 	default:
 		return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("forge: call has no operation variant set"))
 	}
@@ -259,6 +257,68 @@ func (s *forgeService) resolveTarget(call *compassv1internal.ForgeCallRequest, r
 	return rf, nil
 }
 
+// subscribeToStoreKind maps the wire ForgeArtifactKind onto the store enum,
+// rejecting UNSPECIFIED(0) as an in-band invalid_argument (a subscription must
+// name an issue or a pull request). The two enums share their numeric domain
+// (issue=1, pull_request=2), so a known kind is a direct cast.
+func subscribeToStoreKind(kind compassv1internal.ForgeArtifactKind) (store.ForgeArtifactKind, *compassv1internal.ForgeCallError) {
+	switch kind {
+	case compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE:
+		return store.ForgeArtifactKindIssue, nil
+	case compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_PULL_REQUEST:
+		return store.ForgeArtifactKindPullRequest, nil
+	default:
+		return store.ForgeArtifactKindUnspecified, forgeErr(connect.CodeInvalidArgument, "forge: subscription kind must be issue or pull_request")
+	}
+}
+
+// subscribeForge records the caller's standing interest in one forge artifact
+// (DL-053). It resolves the coordinate (repo + provider/host), maps the wire
+// kind to the store enum, and idempotently ensures the subscription row —
+// returning the EXISTING subscription id on a repeat (the store upsert dedups on
+// the UNIQUE coordinate per agent). No owner stamp: a subscribe authors nothing.
+func (s *forgeService) subscribeForge(ctx context.Context, caller store.AccountID, call *compassv1internal.ForgeCallRequest, req *compassv1internal.SubscribeForgeRequest) *compassv1internal.ForgeCallResult {
+	rf, fe := s.resolveTarget(call, req.GetRepo())
+	if fe != nil {
+		return forgeErrorResult(fe)
+	}
+	kind, fe := subscribeToStoreKind(req.GetKind())
+	if fe != nil {
+		return forgeErrorResult(fe)
+	}
+	id, err := s.store.EnsureAgentForgeSubscription(ctx, store.AgentForgeSubscription{
+		AgentAccountID: caller,
+		Provider:       store.ForgeProvider(rf.provider),
+		Host:           rf.host,
+		Repo:           req.GetRepo(),
+		Kind:           kind,
+		Number:         req.GetNumber(),
+	})
+	if err != nil {
+		return forgeErrorResult(storeForgeError(err))
+	}
+	return &compassv1internal.ForgeCallResult{
+		Result: &compassv1internal.ForgeCallResult_Subscribed{
+			Subscribed: &compassv1internal.SubscribeForgeResponse{SubscriptionId: id},
+		},
+	}
+}
+
+// unsubscribeForge deletes the caller's subscription by id (scoped to the
+// calling agent — an unknown id, or one owned by another agent, is an in-band
+// not_found), running the DL-053 last-subscription cursor GC in the store's
+// transaction. Unsubscribe is by id, so it needs no coordinate resolution.
+func (s *forgeService) unsubscribeForge(ctx context.Context, caller store.AccountID, req *compassv1internal.UnsubscribeForgeRequest) *compassv1internal.ForgeCallResult {
+	if err := s.store.DeleteAgentForgeSubscription(ctx, caller, req.GetSubscriptionId()); err != nil {
+		return forgeErrorResult(storeForgeError(err))
+	}
+	return &compassv1internal.ForgeCallResult{
+		Result: &compassv1internal.ForgeCallResult_Unsubscribed{
+			Unsubscribed: &compassv1internal.UnsubscribeForgeResponse{},
+		},
+	}
+}
+
 // dedup is the F3 idempotency-memo lookup: the artifact the agent authored under
 // clientRequestID, or ok=false on a miss. An empty key is always a miss (it is
 // never stored), short-circuited so it never touches the store.
diff --git a/go/server/forge_test.go b/go/server/forge_test.go
index e3a886d7..b8a1da21 100644
--- a/go/server/forge_test.go
+++ b/go/server/forge_test.go
@@ -21,6 +21,7 @@ package server
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -51,15 +52,66 @@ type fakeForgeStore struct {
 	recorded []store.AuthoredArtifact
 	getErr   error // if set, GetAccount returns it verbatim
 	recErr   error // if set, RecordAuthoredArtifact returns it verbatim
+
+	// 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
+	// let a test force a store fault on either path.
+	subs    map[string]store.AgentForgeSubscription
+	subKey  map[string]string // agent|provider|host|repo|kind|number -> id
+	nextSub int
+	subErr  error
+	delErr  error
 }
 
 func newFakeForgeStore() *fakeForgeStore {
 	return &fakeForgeStore{
 		accounts: make(map[store.AccountID]store.Account),
 		memo:     make(map[string]store.AuthoredArtifact),
+		subs:     make(map[string]store.AgentForgeSubscription),
+		subKey:   make(map[string]string),
 	}
 }
 
+// subCoordKey builds the UNIQUE (agent, coordinate) index key the real store's
+// ON CONFLICT constrains on, so a repeat subscribe re-lands on the same id.
+func subCoordKey(agent store.AccountID, sub store.AgentForgeSubscription) string {
+	return fmt.Sprintf("%s|%d|%s|%s|%d|%d", agent, sub.Provider, sub.Host, sub.Repo, sub.Kind, sub.Number)
+}
+
+// EnsureAgentForgeSubscription mirrors the real store's idempotent upsert: a
+// repeat (agent, coordinate) returns the stored id; a new coordinate mints one.
+func (f *fakeForgeStore) EnsureAgentForgeSubscription(_ context.Context, sub store.AgentForgeSubscription) (string, error) {
+	if f.subErr != nil {
+		return "", f.subErr
+	}
+	key := subCoordKey(sub.AgentAccountID, sub)
+	if id, ok := f.subKey[key]; ok {
+		return id, nil
+	}
+	f.nextSub++
+	id := fmt.Sprintf("sub-%d", f.nextSub)
+	sub.ID = id
+	f.subs[id] = sub
+	f.subKey[key] = id
+	return id, nil
+}
+
+// DeleteAgentForgeSubscription mirrors the real store's id+agent-scoped delete:
+// an unknown id, or one owned by another agent, is ErrNotFound.
+func (f *fakeForgeStore) DeleteAgentForgeSubscription(_ context.Context, agent store.AccountID, subscriptionID string) error {
+	if f.delErr != nil {
+		return f.delErr
+	}
+	sub, ok := f.subs[subscriptionID]
+	if !ok || sub.AgentAccountID != agent {
+		return store.ErrNotFound
+	}
+	delete(f.subs, subscriptionID)
+	delete(f.subKey, subCoordKey(agent, sub))
+	return nil
+}
+
 func (f *fakeForgeStore) GetAccount(_ context.Context, id store.AccountID) (store.Account, error) {
 	if f.getErr != nil {
 		return store.Account{}, f.getErr
@@ -739,6 +791,150 @@ func TestForgeRecordFailureAfterProviderSuccessIsInternal(t *testing.T) {
 	}
 }
 
+// --- tests: DL-053 subscribe/unsubscribe arms (RIG-2732 Piece 1) -------------
+
+// subscribeCall builds a Subscribe request against the default GitHub coordinate.
+func subscribeCall(kind compassv1internal.ForgeArtifactKind, number uint64) *compassv1internal.ForgeCallRequest {
+	return &compassv1internal.ForgeCallRequest{
+		Call: &compassv1internal.ForgeCallRequest_Subscribe{Subscribe: &compassv1internal.SubscribeForgeRequest{
+			Repo: testRepo, Kind: kind, Number: number,
+		}},
+	}
+}
+
+// unsubscribeCall builds an Unsubscribe request by subscription id.
+func unsubscribeCall(subscriptionID string) *compassv1internal.ForgeCallRequest {
+	return &compassv1internal.ForgeCallRequest{
+		Call: &compassv1internal.ForgeCallRequest_Unsubscribe{Unsubscribe: &compassv1internal.UnsubscribeForgeRequest{
+			SubscriptionId: subscriptionID,
+		}},
+	}
+}
+
+// TestForgeSubscribeReturnsIdAndIsIdempotent pins the subscribe arm: a subscribe
+// returns a non-empty subscription id, and a REPEAT subscribe to the same
+// artifact returns the SAME id (the store upsert dedups on the UNIQUE
+// coordinate) — not a fresh row, not an error.
+func TestForgeSubscribeReturnsIdAndIsIdempotent(t *testing.T) {
+	author := forge.NewFakeProvider("gh-author")
+	reviewer := forge.NewFakeProvider("gh-reviewer")
+	svc, st := newForgeServiceForTest(t, author, reviewer)
+
+	res := svc.ExecuteForgeCallAsAccountMust(t, subscribeCall(compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE, 42))
+	sub := res.GetSubscribed()
+	if sub == nil || sub.GetSubscriptionId() == "" {
+		t.Fatalf("subscribe result = %v, want a subscription id", res.GetResult())
+	}
+	first := sub.GetSubscriptionId()
+
+	res2 := svc.ExecuteForgeCallAsAccountMust(t, subscribeCall(compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE, 42))
+	if got := res2.GetSubscribed().GetSubscriptionId(); got != first {
+		t.Fatalf("repeat subscribe id = %q, want %q (idempotent)", got, first)
+	}
+	if len(st.subs) != 1 {
+		t.Fatalf("subscription rows = %d, want 1 (no duplicate)", len(st.subs))
+	}
+}
+
+// TestForgeSubscribeUnspecifiedKindIsInvalidArgument pins the kind guard: a
+// subscribe with an UNSPECIFIED kind is an in-band invalid_argument with no row
+// written — the arm rejects the zero kind before the store.
+func TestForgeSubscribeUnspecifiedKindIsInvalidArgument(t *testing.T) {
+	author := forge.NewFakeProvider("gh-author")
+	reviewer := forge.NewFakeProvider("gh-reviewer")
+	svc, st := newForgeServiceForTest(t, author, reviewer)
+
+	res := svc.ExecuteForgeCallAsAccountMust(t, subscribeCall(compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_UNSPECIFIED, 1))
+	if fe := res.GetError(); fe == nil || fe.GetCode() != "invalid_argument" {
+		t.Fatalf("unspecified kind error = %v, want invalid_argument", res.GetError())
+	}
+	if len(st.subs) != 0 {
+		t.Fatalf("subscription rows = %d, want 0", len(st.subs))
+	}
+}
+
+// TestForgeSubscribeEmptyRepoIsInvalidArgument pins that an empty repo is an
+// in-band invalid_argument before any store touch (resolveTarget guard).
+func TestForgeSubscribeEmptyRepoIsInvalidArgument(t *testing.T) {
+	author := forge.NewFakeProvider("gh-author")
+	reviewer := forge.NewFakeProvider("gh-reviewer")
+	svc, st := newForgeServiceForTest(t, author, reviewer)
+
+	call := &compassv1internal.ForgeCallRequest{
+		Call: &compassv1internal.ForgeCallRequest_Subscribe{Subscribe: &compassv1internal.SubscribeForgeRequest{
+			Repo: "", Kind: compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE, Number: 1,
+		}},
+	}
+	res := svc.ExecuteForgeCallAsAccountMust(t, call)
+	if fe := res.GetError(); fe == nil || fe.GetCode() != "invalid_argument" {
+		t.Fatalf("empty repo error = %v, want invalid_argument", res.GetError())
+	}
+	if len(st.subs) != 0 {
+		t.Fatalf("subscription rows = %d, want 0", len(st.subs))
+	}
+}
+
+// TestForgeUnsubscribeSucceedsThenNotFound pins the unsubscribe arm: an existing
+// subscription id unsubscribes to the Unsubscribed arm and removes the row; a
+// repeat (now-unknown) id is an in-band not_found — never a Connect teardown.
+func TestForgeUnsubscribeSucceedsThenNotFound(t *testing.T) {
+	author := forge.NewFakeProvider("gh-author")
+	reviewer := forge.NewFakeProvider("gh-reviewer")
+	svc, st := newForgeServiceForTest(t, author, reviewer)
+
+	sub := svc.ExecuteForgeCallAsAccountMust(t, subscribeCall(compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_PULL_REQUEST, 7)).GetSubscribed()
+	id := sub.GetSubscriptionId()
+
+	res := svc.ExecuteForgeCallAsAccountMust(t, unsubscribeCall(id))
+	if res.GetUnsubscribed() == nil {
+		t.Fatalf("unsubscribe result = %v, want Unsubscribed", res.GetResult())
+	}
+	if len(st.subs) != 0 {
+		t.Fatalf("subscription rows = %d, want 0 after unsubscribe", len(st.subs))
+	}
+
+	again := svc.ExecuteForgeCallAsAccountMust(t, unsubscribeCall(id))
+	if fe := again.GetError(); fe == nil || fe.GetCode() != "not_found" {
+		t.Fatalf("repeat unsubscribe error = %v, want not_found", again.GetError())
+	}
+}
+
+// TestForgeUnsubscribeBogusIdIsInbandNotFound pins that an unknown subscription
+// id is an in-band not_found ForgeCallError, NOT a Connect error.
+func TestForgeUnsubscribeBogusIdIsInbandNotFound(t *testing.T) {
+	author := forge.NewFakeProvider("gh-author")
+	reviewer := forge.NewFakeProvider("gh-reviewer")
+	svc, _ := newForgeServiceForTest(t, author, reviewer)
+
+	res := svc.ExecuteForgeCallAsAccountMust(t, unsubscribeCall("no-such-sub"))
+	if fe := res.GetError(); fe == nil || fe.GetCode() != "not_found" {
+		t.Fatalf("bogus unsubscribe error = %v, want not_found", res.GetError())
+	}
+}
+
+// TestForgeSubscribeUnsubscribeStoreFaultIsInbandInternal pins the store-fault
+// rail: a generic store failure on either the subscribe or unsubscribe path
+// renders as an in-band ForgeCallError (code "internal", via storeForgeError),
+// never a Connect stream teardown — the same tool-failure-is-not-a-teardown
+// contract the read/write arms hold.
+func TestForgeSubscribeUnsubscribeStoreFaultIsInbandInternal(t *testing.T) {
+	author := forge.NewFakeProvider("gh-author")
+	reviewer := forge.NewFakeProvider("gh-reviewer")
+	svc, st := newForgeServiceForTest(t, author, reviewer)
+
+	st.subErr = errors.New("boom: subscribe store fault")
+	subRes := svc.ExecuteForgeCallAsAccountMust(t, subscribeCall(compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE, 42))
+	if fe := subRes.GetError(); fe == nil || fe.GetCode() != "internal" {
+		t.Fatalf("subscribe store-fault error = %v, want in-band internal", subRes.GetError())
+	}
+
+	st.delErr = errors.New("boom: unsubscribe store fault")
+	delRes := svc.ExecuteForgeCallAsAccountMust(t, unsubscribeCall("any-id"))
+	if fe := delRes.GetError(); fe == nil || fe.GetCode() != "internal" {
+		t.Fatalf("unsubscribe store-fault error = %v, want in-band internal", delRes.GetError())
+	}
+}
+
 // --- small helpers ----------------------------------------------------------
 
 func itoa(n int) string {