MUL-7051 fix(subscribers): stop an armed autopilot from subscribing its trigger's creator (#8031)

* fix(subscribers): stop an armed autopilot from subscribing its trigger's creator

An autopilot fires on its schedule, its agent files issues, and the member who
armed the trigger is auto-subscribed to every one of them.

The delegated-subscriber rule (MUL-5483) subscribes the human an agent-created
issue exists on behalf of, and it tested that by asking whether the origin run
carries an originator. That was sound when it was written: direct_human was the
only attribution root that left an originator behind, so "this run carries a
human" and "a human asked for this work" were the same statement.

MUL-6951 separated them. An armed schedule/webhook trigger now runs with its
creator's authorization (trigger_owner), and attribution COPIES that human down
every agent hop, so the rule started reading an automation's authorization as a
request. attribution.DelegatedSubscriber already documents this exact outcome as
one it must not produce — the origin_type='autopilot' exclusion says the
autopilot's own subscriber template, not whoever armed the trigger, is the
audience for its issues. The regression arrives one hop below that exclusion,
where the issue's origin is an ordinary agent_create.

The rule now asks what it means: did this chain BEGIN with a member acting. The
originator alone cannot answer that below the first hop, because a hop relabels
its own originator_source to 'delegation' — so GetDelegatedSubscriptionFacts
walks delegated_from_task_id back to the run that actually resolved the human and
returns that root's label alongside the originator, in one workspace-scoped read
that replaces the previous single-task load. Accepting only direct_human makes it
a whitelist: the next root that resolves a human some other way stays silent here
until someone decides otherwise, instead of switching this rule on for a
population nobody chose.

No schema change and no change to attribution itself: originator, accountable and
every authorization decision are untouched. Only who gets subscribed moves.

Verification (real PostgreSQL 17): internal/attribution, cmd/server pass; the
four new tests fail against the unfixed rule, and the tenant-guard test fails
when the workspace predicate is removed from the new query. internal/handler and
internal/service each keep one pre-existing failure that reproduces identically
on a clean tree (source-context object-store cleanup), as do cmd/multica,
internal/cli, internal/daemon and internal/daemon/execenv.

Co-authored-by: multica-agent <[email protected]>

* test(subscribers): close the review gaps in the lineage walk's coverage

Addresses Elon's review on #8031.

The 32-hop stop is accepted as a product limit (Bohan's call), so it is now
written down as behaviour instead of contradicted by it. DelegatedSubscriber
claimed "No depth cap", which was true of WHO the human is — the originator is
copied onto every run — but not of proving the chain began with them, which is
the part that walks. The query comment called the cap cycle protection only. Both
now say what actually happens: past 32 hops the root is unproven and nobody is
subscribed. TestDelegatedSubscribe_LineageWalkStopsAt32Hops pins both sides, so
the limit cannot drift silently; moving it to 3 reddens the 32-hop case.

The tenant test claimed to cover every hop and covered one. An issue pointing
straight at a foreign run only exercises the anchor's predicate, so deleting the
recursive branch's workspace_id left it green. It is now two cases, and the added
one — a local run naming a foreign parent — fails with that predicate removed
while the original case still passes, which is exactly the blind spot.

The chain fixture builds through internal/testutil instead of hand-writing an
INSERT with a matching cleanup, per the repository's test convention.

No production behaviour changes: the rule, the walk and every existing assertion
are untouched.

Verification (real PostgreSQL 17): cmd/server, internal/attribution,
internal/service pass; both new tests were run against a deliberately broken
query and fail there. gofmt, go vet, git diff --check clean.

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: J <[email protected]>

server/cmd/server/delegated_subscriber_membership_test.go

@@ -43,8 +43,8 @@ func addWorkspaceMember(t *testing.T, userID string) (revoke func()) {
 // round 6 finding 2.
 //
 // originator_user_id is stamped when the task is QUEUED and never revisited.
-// GetAgentTaskInWorkspace proves the task's agent still belongs to the
-// workspace, which is a different claim from "the human does". A member can be
+// The lineage read proves the task's agent still belongs to the workspace,
+// which is a different claim from "the human does". A member can be
 // revoked while their tasks keep running on a shared runtime, and every child
 // those tasks file afterwards would write a subscriber row for a non-member:
 // a ghost watcher accumulating inbox rows, and restored history if they ever

server/cmd/server/delegated_subscriber_test.go

@@ -2,11 +2,14 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/multica-ai/multica/server/internal/events"
 	"github.com/multica-ai/multica/server/internal/handler"
+	"github.com/multica-ai/multica/server/internal/testutil"
 	"github.com/multica-ai/multica/server/internal/util"
 	db "github.com/multica-ai/multica/server/pkg/db/generated"
 	"github.com/multica-ai/multica/server/pkg/protocol"
@@ -29,24 +32,36 @@ func firstFixtureAgent(t *testing.T) (agentID, runtimeID string) {
 }
 
 // createAgentTaskWithOriginator inserts a queued task attributed to the given
-// human, standing in for the run an agent is executing when it files an issue.
-// An invalid originator models an unattributed / autopilot-rooted chain.
+// human by that human's own action, standing in for the run an agent is
+// executing when it files an issue. An invalid originator models an
+// unattributed chain.
 func createAgentTaskWithOriginator(t *testing.T, agentID, runtimeID string, originator pgtype.UUID) pgtype.UUID {
 	t.Helper()
-	ctx := context.Background()
-	var taskID pgtype.UUID
-	err := testPool.QueryRow(ctx, `
-		INSERT INTO agent_task_queue (agent_id, runtime_id, status, priority, originator_user_id, accountable_user_id)
-		VALUES ($1, $2, 'queued', 0, $3, $3)
-		RETURNING id
-	`, agentID, runtimeID, originator).Scan(&taskID)
-	if err != nil {
-		t.Fatalf("create agent task: %v", err)
-	}
-	t.Cleanup(func() {
-		testPool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE id = $1`, taskID)
-	})
-	return taskID
+	return createAgentTaskInChain(t, agentID, runtimeID, originator, "direct_human", pgtype.UUID{})
+}
+
+// createAgentTaskInChain is the same insert with the attribution lineage spelled
+// out: the waterfall level that resolved the human, and the parent run the human
+// was copied from on a delegation hop. Both matter to the delegated rule since
+// MUL-7051 — the originator alone no longer says whether a human asked for the
+// work, only that the run carries one.
+func createAgentTaskInChain(t *testing.T, agentID, runtimeID string, originator pgtype.UUID, source string, delegatedFrom pgtype.UUID) pgtype.UUID {
+	t.Helper()
+	return util.MustParseUUID(workspaceFixture(t).Task(t, agentID, testutil.Cols{
+		"runtime_id":             runtimeID,
+		"originator_user_id":     originator,
+		"accountable_user_id":    originator,
+		"originator_source":      source,
+		"delegated_from_task_id": delegatedFrom,
+	}))
+}
+
+// workspaceFixture builds rows in the suite's own workspace. Every row it
+// inserts is removed when the test that asked for it ends, which is why the
+// helpers above no longer pair an INSERT with a hand-written cleanup.
+func workspaceFixture(t *testing.T) *testutil.Fixture {
+	t.Helper()
+	return testutil.New(testPool, testWorkspaceID, testUserID)
 }
 
 // createAgentOriginIssue inserts an issue whose creator is an agent and whose
@@ -171,6 +186,179 @@ func TestDelegatedSubscribe_UnattributedOriginSubscribesNobody(t *testing.T) {
 	}
 }
 
+// TestDelegatedSubscribe_ArmedTriggerRunSubscribesNobody is MUL-7051 as the user
+// met it: an autopilot fires on its schedule, its agent files issues all day,
+// and the member who armed the trigger months ago is subscribed to every one.
+//
+// The task row here is the shape MUL-6951 introduced and the reason a unit test
+// of the rule is not enough — a real trigger_owner run carries a real member as
+// originator, so nothing about the row it hands the listener looks degraded.
+func TestDelegatedSubscribe_ArmedTriggerRunSubscribesNobody(t *testing.T) {
+	queries := db.New(testPool)
+	bus := events.New()
+	registerSubscriberListeners(bus, testPool)
+
+	agentID, runtimeID := firstFixtureAgent(t)
+	task := createAgentTaskInChain(t, agentID, runtimeID, util.MustParseUUID(testUserID), "trigger_owner", pgtype.UUID{})
+	issueID := createAgentOriginIssue(t, agentID, "agent_create", task, pgtype.UUID{})
+
+	publishAgentIssueCreated(bus, issueID, agentID)
+
+	if isSubscribed(t, queries, issueID, "member", testUserID) {
+		t.Fatal("an issue filed inside an autopilot run must not subscribe the member who armed the trigger")
+	}
+}
+
+// TestDelegatedSubscribe_ChainRootDecidesBelowTheFirstHop: the fix cannot read
+// the origin run's own label, because a hop overwrites it with 'delegation'. One
+// agent handing work to another is the ordinary way an autopilot run grows, and
+// at that depth the delegated run is indistinguishable from a human-triggered
+// one by anything except the chain it came from.
+//
+// Both directions are asserted together: the same shape, differing only in what
+// started the chain, must reach opposite answers — otherwise a fix that simply
+// stopped subscribing below the first hop would pass.
+func TestDelegatedSubscribe_ChainRootDecidesBelowTheFirstHop(t *testing.T) {
+	queries := db.New(testPool)
+	bus := events.New()
+	registerSubscriberListeners(bus, testPool)
+
+	agentID, runtimeID := firstFixtureAgent(t)
+	human := util.MustParseUUID(testUserID)
+
+	for _, tc := range []struct {
+		name string
+		root string
+		want bool
+	}{
+		{"member asked, two agents deep", "direct_human", true},
+		{"schedule fired, two agents deep", "trigger_owner", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			// The originator is copied, not chained, so it is the SAME member on
+			// both runs either way (MUL-4302 §3.2).
+			root := createAgentTaskInChain(t, agentID, runtimeID, human, tc.root, pgtype.UUID{})
+			hop := createAgentTaskInChain(t, agentID, runtimeID, human, "delegation", root)
+			issueID := createAgentOriginIssue(t, agentID, "agent_create", hop, pgtype.UUID{})
+
+			publishAgentIssueCreated(bus, issueID, agentID)
+
+			if got := isSubscribed(t, queries, issueID, "member", testUserID); got != tc.want {
+				t.Fatalf("subscribed = %v, want %v for a chain rooted at %s", got, tc.want, tc.root)
+			}
+		})
+	}
+}
+
+// TestDelegatedSubscribe_LineageWalkStopsAt32Hops pins the one place the fix is
+// not exact. Resolving WHO the human is has no depth limit — the originator is
+// copied onto every run — but proving the chain BEGAN with them means walking to
+// the root, and that walk stops after 32 hops. A member-rooted chain longer than
+// that loses its subscription.
+//
+// The limit is accepted (MUL-7051), so it is written down as behaviour rather
+// than left as an implementation detail of the query: an accepted limit nothing
+// asserts is one that moves by accident. Both sides are pinned, because a test
+// for the truncation alone would still pass if the walk stopped at hop 3.
+func TestDelegatedSubscribe_LineageWalkStopsAt32Hops(t *testing.T) {
+	queries := db.New(testPool)
+	bus := events.New()
+	registerSubscriberListeners(bus, testPool)
+
+	agentID, runtimeID := firstFixtureAgent(t)
+	human := util.MustParseUUID(testUserID)
+
+	// hops is the distance from the run that files the issue up to the root.
+	buildChain := func(t *testing.T, hops int) pgtype.UUID {
+		t.Helper()
+		run := createAgentTaskInChain(t, agentID, runtimeID, human, "direct_human", pgtype.UUID{})
+		for i := 0; i < hops; i++ {
+			run = createAgentTaskInChain(t, agentID, runtimeID, human, "delegation", run)
+		}
+		return run
+	}
+
+	for _, tc := range []struct {
+		name string
+		hops int
+		want bool
+	}{
+		{"the deepest chain still proven", 32, true},
+		{"one hop past the limit", 33, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			issueID := createAgentOriginIssue(t, agentID, "agent_create", buildChain(t, tc.hops), pgtype.UUID{})
+
+			publishAgentIssueCreated(bus, issueID, agentID)
+
+			if got := isSubscribed(t, queries, issueID, "member", testUserID); got != tc.want {
+				t.Fatalf("subscribed = %v, want %v at %d hops from the root", got, tc.want, tc.hops)
+			}
+		})
+	}
+}
+
+// TestDelegatedSubscribe_ForeignLineageResolvesNobody pins the tenant guard
+// (MUL-4252) across the read that now performs it. origin_id and
+// delegated_from_task_id are both bare UUIDs with nothing scoping them, so an id
+// belonging to another workspace's run must resolve no human at all — not the
+// one that run carries.
+//
+// Both hops are covered because they are separate predicates in the recursive
+// query, and only the anchor's is exercised by an issue pointing straight at a
+// foreign run. A chain that starts here and STEPS OUT one hop up is what proves
+// the recursive branch carries the guard too: with it, the lineage stops at the
+// local hop and reports 'delegation' — unproven, so nobody is subscribed; without
+// it, the walk reads a foreign root, sees direct_human, and subscribes.
+func TestDelegatedSubscribe_ForeignLineageResolvesNobody(t *testing.T) {
+	queries := db.New(testPool)
+	bus := events.New()
+	registerSubscriberListeners(bus, testPool)
+
+	// A whole run in a workspace this issue has nothing to do with, attributed
+	// to a member of THIS one — the shape a foreign id would exploit.
+	slug := fmt.Sprintf("mul7051-foreign-%d", time.Now().UnixNano())
+	unbound := testutil.New(testPool, "", testUserID)
+	foreign := testutil.New(testPool, unbound.Workspace(t, "MUL-7051 foreign", slug), testUserID)
+	foreignRuntime := foreign.Runtime(t, slug+"-runtime")
+	foreignAgent := foreign.Agent(t, slug+"-agent", foreignRuntime)
+	foreignRoot := util.MustParseUUID(foreign.Task(t, foreignAgent, testutil.Cols{
+		"runtime_id":          foreignRuntime,
+		"originator_user_id":  testUserID,
+		"accountable_user_id": testUserID,
+		"originator_source":   "direct_human",
+	}))
+
+	agentID, runtimeID := firstFixtureAgent(t)
+
+	for _, tc := range []struct {
+		name   string
+		origin func() pgtype.UUID
+	}{
+		{
+			name:   "the issue points straight at a foreign run",
+			origin: func() pgtype.UUID { return foreignRoot },
+		},
+		{
+			name: "a local run claims a foreign parent",
+			origin: func() pgtype.UUID {
+				return createAgentTaskInChain(t, agentID, runtimeID,
+					util.MustParseUUID(testUserID), "delegation", foreignRoot)
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			issueID := createAgentOriginIssue(t, agentID, "agent_create", tc.origin(), pgtype.UUID{})
+
+			publishAgentIssueCreated(bus, issueID, agentID)
+
+			if isSubscribed(t, queries, issueID, "member", testUserID) {
+				t.Fatal("a lineage that leaves the workspace must not resolve a subscriber")
+			}
+		})
+	}
+}
+
 // TestDelegatedSubscribe_RespectsSubtreeOptOut is the reason the tombstone
 // exists. Without it "stop watching this tree" is undone by the very next child
 // the agent files under it: the rule fires per issue, and an agent-built tree

server/cmd/server/subscriber_listeners.go

@@ -152,6 +152,12 @@ func registerSubscriberListeners(bus *events.Bus, pool *pgxpool.Pool) {
 // rule is the point: the defect being fixed is attribution and notification
 // disagreeing about whose behalf an issue exists on.
 //
+// Attribution answers "whose authority does this run carry", which since MUL-6951
+// is a broader population than "who asked for this work" — an armed autopilot
+// trigger carries its creator's. Visibility follows the narrower one, so the read
+// also walks the chain back to the run that resolved the human and hands the rule
+// that root's label (MUL-7051).
+//
 // Everything is best-effort and logged, never fatal — the issue is already
 // committed and a subscription hiccup must not look like a creation failure.
 func subscribeDelegatedHuman(bus *events.Bus, pool *pgxpool.Pool, queries *db.Queries, workspaceID, issueID string) {
@@ -169,11 +175,13 @@ func subscribeDelegatedHuman(bus *events.Bus, pool *pgxpool.Pool, queries *db.Qu
 		return
 	}
 
-	// Workspace-scoped so a foreign origin id can never resolve a human from
-	// another tenant (the MUL-4252 guard the comment chain already applies).
-	originTask, err := queries.GetAgentTaskInWorkspace(ctx, db.GetAgentTaskInWorkspaceParams{
-		ID:          issue.OriginID,
-		WorkspaceID: parseUUID(workspaceID),
+	// The origin run's human, plus how the chain it belongs to acquired that
+	// human — one walk, workspace-scoped at every hop so a foreign origin id can
+	// never resolve someone from another tenant (the MUL-4252 guard the comment
+	// chain already applies).
+	facts, err := queries.GetDelegatedSubscriptionFacts(ctx, db.GetDelegatedSubscriptionFactsParams{
+		OriginTaskID: issue.OriginID,
+		WorkspaceID:  parseUUID(workspaceID),
 	})
 	if err != nil {
 		// A missing origin task is normal (cancelled/reaped run), not an error
@@ -186,7 +194,8 @@ func subscribeDelegatedHuman(bus *events.Bus, pool *pgxpool.Pool, queries *db.Qu
 	human, reason, ok := attribution.DelegatedSubscriber(attribution.SubscriptionFacts{
 		CreatorType:      issue.CreatorType,
 		OriginType:       issue.OriginType.String,
-		OriginOriginator: originTask.OriginatorUserID,
+		OriginOriginator: facts.OriginatorUserID,
+		OriginRootSource: attribution.Source(facts.RootSource.String),
 	})
 	if !ok {
 		return

server/internal/attribution/attribution.go

@@ -453,6 +453,18 @@ type SubscriptionFacts struct {
 	// stamp and the origin task's originator_user_id, loaded by the caller.
 	OriginType       string
 	OriginOriginator pgtype.UUID
+
+	// OriginRootSource is the originator_source of the CHAIN ROOT behind the
+	// origin task — the run that actually resolved OriginOriginator, not the hop
+	// that copied it. The caller walks delegated_from_task_id to find it
+	// (GetDelegatedSubscriptionFacts); an unreachable or unlabelled root leaves
+	// it empty, which the rule reads as "not a direct human act".
+	//
+	// It exists because OriginOriginator alone stopped answering "did a human ask
+	// for this?" in MUL-6951: an armed autopilot trigger now carries its
+	// creator's authorization, so the two cases became indistinguishable by
+	// value. See DelegatedSubscriber.
+	OriginRootSource Source
 }
 
 // DelegatedSubscriber resolves the human who should be auto-subscribed to an
@@ -463,11 +475,21 @@ type SubscriptionFacts struct {
 // attribution and notification disagreeing about that (MUL-5483). The waterfall
 // is narrower than ClassifyDirect's on purpose:
 //
-//   - OriginOriginator valid → subscribe. A human authorized this chain, and
-//     because attribution COPIES the accountable human across every agent hop
-//     rather than chaining it (see SourceDelegation), this resolves the ORIGINAL
-//     human at any depth. No depth cap: depth is exactly where a lost signal
-//     hurts most, and the delivery tier — not a cap — is what bounds the noise.
+//   - OriginOriginator valid AND the chain root is a DIRECT human act →
+//     subscribe. Because attribution COPIES the accountable human across every
+//     agent hop rather than chaining it (see SourceDelegation), the originator
+//     resolves the ORIGINAL human at any depth, and the delivery tier — not a
+//     depth cutoff — is what bounds the noise a deep chain makes. The root check
+//     is what keeps "has a human" from being read as "a human asked", see below.
+//
+//     WHO the human is therefore has no depth limit; proving the chain BEGAN
+//     with them does: the caller reads the root by walking the lineage, and that
+//     walk stops after 32 hops (GetDelegatedSubscriptionFacts). Past that the
+//     root is reported as unproven and nobody is subscribed, which is a real
+//     truncation and not a formality — accepted deliberately (MUL-7051, Bohan's
+//     call) because no observed chain approaches it. Raising it is a one-number
+//     change; moving the root onto the run at enqueue time would remove the
+//     limit outright.
 //
 //   - quick_create → reason 'creator', agent_create → reason 'delegated'. The
 //     quick-create human asked for THAT issue by name, so it is direct intent and
@@ -480,10 +502,29 @@ type SubscriptionFacts struct {
 //     to arm the trigger — is the intended audience for its issues. Degraded
 //     attribution (owner_fallback / unattributed) is excluded for the same
 //     reason it is degraded: we do not fabricate a human to notify.
+//
+// WHY the root source is checked at all (MUL-7051). This rule was written when
+// SourceDirectHuman was the only root that left an originator behind, so "the
+// origin run carries a human" and "a human asked for this work" were the same
+// statement and only the first had to be tested. MUL-6951 separated them: an
+// armed schedule/webhook trigger now runs with its creator's authorization
+// (SourceTriggerOwner), and that human is copied down the whole chain exactly
+// like a requester would be. The untested half of the old equivalence is what
+// broke — every issue an autopilot's agent filed started subscribing whoever
+// armed the trigger, which is the case the origin_type='autopilot' exclusion
+// above already says must not happen, arriving one hop lower.
+//
+// So the condition is stated as what it means — the chain BEGAN with a member
+// acting — and it is a whitelist. A future root that resolves a human some other
+// way stays silent here until someone decides it should subscribe people, rather
+// than turning this rule on for a population nobody chose.
 func DelegatedSubscriber(f SubscriptionFacts) (pgtype.UUID, string, bool) {
 	if f.CreatorType != "agent" || !f.OriginOriginator.Valid {
 		return pgtype.UUID{}, "", false
 	}
+	if f.OriginRootSource != SourceDirectHuman {
+		return pgtype.UUID{}, "", false
+	}
 	switch f.OriginType {
 	case "quick_create":
 		return f.OriginOriginator, "creator", true

server/internal/attribution/delegated_subscriber_test.go

@@ -14,6 +14,7 @@ func TestDelegatedSubscriber_AgentCreateInheritsHuman(t *testing.T) {
 		CreatorType:      "agent",
 		OriginType:       "agent_create",
 		OriginOriginator: human,
+		OriginRootSource: SourceDirectHuman,
 	})
 
 	if !ok {
@@ -36,6 +37,7 @@ func TestDelegatedSubscriber_QuickCreateIsDirectIntent(t *testing.T) {
 		CreatorType:      "agent",
 		OriginType:       "quick_create",
 		OriginOriginator: human,
+		OriginRootSource: SourceDirectHuman,
 	})
 
 	if !ok {
@@ -58,6 +60,7 @@ func TestDelegatedSubscriber_AutopilotExcluded(t *testing.T) {
 		CreatorType:      "agent",
 		OriginType:       "autopilot",
 		OriginOriginator: human,
+		OriginRootSource: SourceTriggerOwner,
 	}); ok {
 		t.Fatal("autopilot issues must not take a delegated subscription")
 	}
@@ -71,6 +74,7 @@ func TestDelegatedSubscriber_NoHumanNoSubscription(t *testing.T) {
 		CreatorType:      "agent",
 		OriginType:       "agent_create",
 		OriginOriginator: pgtype.UUID{},
+		OriginRootSource: SourceDirectHuman,
 	}); ok {
 		t.Fatal("an unattributed origin task must not produce a subscriber")
 	}
@@ -84,6 +88,7 @@ func TestDelegatedSubscriber_MemberCreatedIssueUnaffected(t *testing.T) {
 		CreatorType:      "member",
 		OriginType:       "agent_create",
 		OriginOriginator: human,
+		OriginRootSource: SourceDirectHuman,
 	}); ok {
 		t.Fatal("member-created issues must not take a delegated subscription")
 	}
@@ -97,7 +102,46 @@ func TestDelegatedSubscriber_UnknownOriginIsInert(t *testing.T) {
 		CreatorType:      "agent",
 		OriginType:       "some_future_origin",
 		OriginOriginator: human,
+		OriginRootSource: SourceDirectHuman,
 	}); ok {
 		t.Fatal("an unrecognized origin_type must not subscribe anyone")
 	}
 }
+
+// TestDelegatedSubscriber_ArmedTriggerChainSubscribesNobody is MUL-7051. The
+// originator is a real, valid member — the person who armed the schedule — and
+// before MUL-6951 no such value existed, so "there is a human here" was safe to
+// read as "a human asked for this". It no longer is, and every issue the
+// autopilot's agent filed started subscribing that member: the exact outcome
+// TestDelegatedSubscriber_AutopilotExcluded forbids, one hop further down.
+func TestDelegatedSubscriber_ArmedTriggerChainSubscribesNobody(t *testing.T) {
+	if _, _, ok := DelegatedSubscriber(SubscriptionFacts{
+		CreatorType:      "agent",
+		OriginType:       "agent_create",
+		OriginOriginator: human,
+		OriginRootSource: SourceTriggerOwner,
+	}); ok {
+		t.Fatal("an issue filed inside an autopilot run must not subscribe whoever armed the trigger")
+	}
+}
+
+// TestDelegatedSubscriber_UnknownRootIsInert is the origin-vocabulary guard
+// (TestDelegatedSubscriber_UnknownOriginIsInert) applied to the other open
+// vocabulary this rule now reads. MUL-7051 happened because a new root source
+// turned the rule on for a population nobody had chosen; a whitelist means the
+// next one has to be chosen deliberately.
+//
+// An empty label — a pre-attribution row, or a lineage truncated before its root
+// — is the same answer: unproven, so no subscription.
+func TestDelegatedSubscriber_UnknownRootIsInert(t *testing.T) {
+	for _, root := range []Source{SourceDelegation, SourceCommentSource, SourceRuleOwner, SourceOwnerFallback, SourceBackfill, SourceUnattributed, "some_future_root", ""} {
+		if _, _, ok := DelegatedSubscriber(SubscriptionFacts{
+			CreatorType:      "agent",
+			OriginType:       "agent_create",
+			OriginOriginator: human,
+			OriginRootSource: root,
+		}); ok {
+			t.Fatalf("root source %q must not subscribe anyone: only a direct human act is a request", root)
+		}
+	}
+}

server/pkg/db/generated/subscriber.sql.go

@@ -161,6 +161,88 @@ func (q *Queries) DeleteSubscriptionsByMember(ctx context.Context, arg DeleteSub
 	return err
 }
 
+const getDelegatedSubscriptionFacts = `-- name: GetDelegatedSubscriptionFacts :one
+WITH RECURSIVE lineage AS (
+    SELECT atq.id, atq.originator_user_id, atq.originator_source, atq.delegated_from_task_id, 0 AS depth
+    FROM agent_task_queue atq
+    JOIN agent a ON a.id = atq.agent_id
+    WHERE atq.id = $1 AND a.workspace_id = $2
+
+    UNION ALL
+
+    SELECT parent.id, parent.originator_user_id, parent.originator_source,
+           parent.delegated_from_task_id, child.depth + 1
+    FROM lineage child
+    JOIN agent_task_queue parent ON parent.id = child.delegated_from_task_id
+    JOIN agent parent_agent ON parent_agent.id = parent.agent_id
+    WHERE child.originator_source IN ('delegation', 'comment_source')
+      AND parent_agent.workspace_id = $2
+      AND child.depth < 32
+),
+chain_root AS (
+    SELECT originator_source FROM lineage ORDER BY depth DESC LIMIT 1
+)
+SELECT origin.originator_user_id, chain_root.originator_source AS root_source
+FROM lineage origin
+CROSS JOIN chain_root
+WHERE origin.depth = 0
+`
+
+type GetDelegatedSubscriptionFactsParams struct {
+	OriginTaskID pgtype.UUID `json:"origin_task_id"`
+	WorkspaceID  pgtype.UUID `json:"workspace_id"`
+}
+
+type GetDelegatedSubscriptionFactsRow struct {
+	OriginatorUserID pgtype.UUID `json:"originator_user_id"`
+	RootSource       pgtype.Text `json:"root_source"`
+}
+
+// Loads the two facts the delegated-subscriber rule needs about the run that
+// filed an issue: the human that run carried, and HOW the delegation chain the
+// run belongs to acquired that human (MUL-5483, MUL-7051). One walk, so one
+// round trip and one tenant guard rather than two.
+//
+// WHY the second fact is needed. Attribution COPIES the originator across every
+// agent hop instead of chaining it (MUL-4302 §3.2), so originator_user_id alone
+// cannot say why a chain has a human: a member's own comment and an autopilot
+// trigger somebody armed months ago leave an identical value on every
+// descendant run. Until MUL-6951 the difference did not have to be read —
+// direct_human was the only root that left an originator at all, so "has an
+// originator" WAS "a human asked for this". An armed trigger now carries its
+// creator's authorization too, and that equivalence is what silently turned the
+// rule into "subscribe whoever armed the automation" (MUL-7051).
+//
+// Only originator_source separates the two, and every hop relabels itself
+// 'delegation' / 'comment_source' — so it is the chain ROOT's label that has to
+// be read. delegated_from_task_id is the link back to the run a hop copied its
+// human from: follow it while the label is still a hop label and the walk lands
+// on the run that actually resolved the human.
+//
+// Tenancy: every hop re-joins agent and re-checks workspace_id — the same guard
+// GetAgentTaskInWorkspace applies to a single task (MUL-4252) — so a foreign
+// delegated_from_task_id can never pull a row from another tenant. Both hops
+// carry it: the anchor's predicate is the one an issue pointing straight at a
+// foreign run exercises, the recursive one is what stops a local run from
+// claiming a parent outside the workspace.
+//
+// The 32-hop stop is a REAL product limit, not just cycle protection (a run can
+// only reference an older run, so the lineage cannot cycle anyway): a chain
+// deeper than that reports the hop label it stopped on and its human is not
+// subscribed. Accepted deliberately — no observed chain comes close, and this
+// read is on the issue-create path — see DelegatedSubscriber for the contract
+// and what removing the limit would take.
+//
+// A lineage truncated by a deleted ancestor, or by that limit, reports the hop
+// label it stopped on, never a root label, so the caller reads it as "not
+// proven" instead of assuming a human asked.
+func (q *Queries) GetDelegatedSubscriptionFacts(ctx context.Context, arg GetDelegatedSubscriptionFactsParams) (GetDelegatedSubscriptionFactsRow, error) {
+	row := q.db.QueryRow(ctx, getDelegatedSubscriptionFacts, arg.OriginTaskID, arg.WorkspaceID)
+	var i GetDelegatedSubscriptionFactsRow
+	err := row.Scan(&i.OriginatorUserID, &i.RootSource)
+	return i, err
+}
+
 const hasAncestorOptOut = `-- name: HasAncestorOptOut :one
 WITH RECURSIVE ancestors(node_id, parent_id, depth) AS (
     SELECT root.id, root.parent_issue_id, 0 FROM issue root WHERE root.id = $1

server/pkg/db/queries/subscriber.sql

@@ -240,3 +240,67 @@ SELECT EXISTS(
     WHERE issue_id = $1 AND user_type = $2 AND user_id = $3
       AND unsubscribed_at IS NULL
 ) AS subscribed;
+
+-- name: GetDelegatedSubscriptionFacts :one
+-- Loads the two facts the delegated-subscriber rule needs about the run that
+-- filed an issue: the human that run carried, and HOW the delegation chain the
+-- run belongs to acquired that human (MUL-5483, MUL-7051). One walk, so one
+-- round trip and one tenant guard rather than two.
+--
+-- WHY the second fact is needed. Attribution COPIES the originator across every
+-- agent hop instead of chaining it (MUL-4302 §3.2), so originator_user_id alone
+-- cannot say why a chain has a human: a member's own comment and an autopilot
+-- trigger somebody armed months ago leave an identical value on every
+-- descendant run. Until MUL-6951 the difference did not have to be read —
+-- direct_human was the only root that left an originator at all, so "has an
+-- originator" WAS "a human asked for this". An armed trigger now carries its
+-- creator's authorization too, and that equivalence is what silently turned the
+-- rule into "subscribe whoever armed the automation" (MUL-7051).
+--
+-- Only originator_source separates the two, and every hop relabels itself
+-- 'delegation' / 'comment_source' — so it is the chain ROOT's label that has to
+-- be read. delegated_from_task_id is the link back to the run a hop copied its
+-- human from: follow it while the label is still a hop label and the walk lands
+-- on the run that actually resolved the human.
+--
+-- Tenancy: every hop re-joins agent and re-checks workspace_id — the same guard
+-- GetAgentTaskInWorkspace applies to a single task (MUL-4252) — so a foreign
+-- delegated_from_task_id can never pull a row from another tenant. Both hops
+-- carry it: the anchor's predicate is the one an issue pointing straight at a
+-- foreign run exercises, the recursive one is what stops a local run from
+-- claiming a parent outside the workspace.
+--
+-- The 32-hop stop is a REAL product limit, not just cycle protection (a run can
+-- only reference an older run, so the lineage cannot cycle anyway): a chain
+-- deeper than that reports the hop label it stopped on and its human is not
+-- subscribed. Accepted deliberately — no observed chain comes close, and this
+-- read is on the issue-create path — see DelegatedSubscriber for the contract
+-- and what removing the limit would take.
+--
+-- A lineage truncated by a deleted ancestor, or by that limit, reports the hop
+-- label it stopped on, never a root label, so the caller reads it as "not
+-- proven" instead of assuming a human asked.
+WITH RECURSIVE lineage AS (
+    SELECT atq.id, atq.originator_user_id, atq.originator_source, atq.delegated_from_task_id, 0 AS depth
+    FROM agent_task_queue atq
+    JOIN agent a ON a.id = atq.agent_id
+    WHERE atq.id = sqlc.arg(origin_task_id) AND a.workspace_id = sqlc.arg(workspace_id)
+
+    UNION ALL
+
+    SELECT parent.id, parent.originator_user_id, parent.originator_source,
+           parent.delegated_from_task_id, child.depth + 1
+    FROM lineage child
+    JOIN agent_task_queue parent ON parent.id = child.delegated_from_task_id
+    JOIN agent parent_agent ON parent_agent.id = parent.agent_id
+    WHERE child.originator_source IN ('delegation', 'comment_source')
+      AND parent_agent.workspace_id = sqlc.arg(workspace_id)
+      AND child.depth < 32
+),
+chain_root AS (
+    SELECT originator_source FROM lineage ORDER BY depth DESC LIMIT 1
+)
+SELECT origin.originator_user_id, chain_root.originator_source AS root_source
+FROM lineage origin
+CROSS JOIN chain_root
+WHERE origin.depth = 0;