diff --git a/server/cmd/server/delegated_subscriber_membership_test.go b/server/cmd/server/delegated_subscriber_membership_test.go index 3829396c6..2bee2902d 100644 --- a/server/cmd/server/delegated_subscriber_membership_test.go +++ b/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 diff --git a/server/cmd/server/delegated_subscriber_test.go b/server/cmd/server/delegated_subscriber_test.go index d8e6a2ff9..fbeaaf58e 100644 --- a/server/cmd/server/delegated_subscriber_test.go +++ b/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 diff --git a/server/cmd/server/subscriber_listeners.go b/server/cmd/server/subscriber_listeners.go index afaf05682..3f9b8d084 100644 --- a/server/cmd/server/subscriber_listeners.go +++ b/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 diff --git a/server/internal/attribution/attribution.go b/server/internal/attribution/attribution.go index 20374efe4..c6581925d 100644 --- a/server/internal/attribution/attribution.go +++ b/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 diff --git a/server/internal/attribution/delegated_subscriber_test.go b/server/internal/attribution/delegated_subscriber_test.go index eee4b67dd..94924a090 100644 --- a/server/internal/attribution/delegated_subscriber_test.go +++ b/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) + } + } +} diff --git a/server/pkg/db/generated/subscriber.sql.go b/server/pkg/db/generated/subscriber.sql.go index d696c398e..5cae8907b 100644 --- a/server/pkg/db/generated/subscriber.sql.go +++ b/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 diff --git a/server/pkg/db/queries/subscriber.sql b/server/pkg/db/queries/subscriber.sql index fe8ce9f9a..612ad11fd 100644 --- a/server/pkg/db/queries/subscriber.sql +++ b/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;