diff --git a/docs/agents.md b/docs/agents.md index 49c4228712..ce6d2aba25 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -1721,7 +1721,7 @@ Event watches and timers accept an optional delivery `channel` and thread `root` Without an address, the subscription runs an independent turn in its Interest's lane. Timers retain their occurrence and expiry budgets. An addressed timer waits without spending an occurrence while destination membership is unconfirmed or absent. `watch_show` reports delivery as held or blocked with a reason; delivery resumes once membership is confirmed. Expiry still applies while held. Membership checks never join channels. The Attention panel shows listening channels separately from the delivery destination. -Event matches wait for confirmation of both listening and destination rosters. Confirmed destination nonmembership discards the match. Addressed wakes refresh the Interest instructions after harness startup and queue waits, immediately before submission, while retaining cancellation checks. +Agent watch writes refuse enabled watches whose explicit listening channels cannot deliver events under the same policy used for event delivery, or whose delivery destination lacks confirmed membership. Disabled watches remain editable and removable. Event matches wait for confirmation of both listening and destination rosters. Confirmed destination nonmembership discards the match. Addressed wakes refresh the Interest instructions after harness startup and queue waits, immediately before submission, while retaining cancellation checks. Conversation wakes are local, have no signed message to resume, and never publish their synthetic anchor ids. Pi saves an Attention wake's reply destination separately from its tool context so a recovered answer keeps its destination. Ordinary plugin wakes retain the separate contract above. diff --git a/docs/buzz-help.md b/docs/buzz-help.md index 04df542872..d3caa37c58 100644 --- a/docs/buzz-help.md +++ b/docs/buzz-help.md @@ -1203,7 +1203,7 @@ confirmation. **Delete team** asks first; it doesn't delete the team's agents. mention can still wake an agent when Attention is off; turning the plugin on does not grant new access. An event watch hears the channels its agent may read, including public channels the agent has not joined. A watch's delivery channel must be one the agent is - a member of; the relay decides where the agent may post. On macOS and Linux, + a member of; the relay decides where the agent may post. Agents cannot enable a watch for an unreadable listening channel or a delivery channel without confirmed membership. On macOS and Linux, watched commands stop when their Buzz process exits. During a handoff between Buzz builds, the new copy waits for the previous command to stop, then continues the watch. diff --git a/docs/relay-queries.md b/docs/relay-queries.md index 92c17ea6c8..903d31b9c6 100644 --- a/docs/relay-queries.md +++ b/docs/relay-queries.md @@ -344,12 +344,12 @@ which it is true when it arrived, kept while the event waits, so watches do not change who a mention wakes; they still hear the rest. A channel leaves the agents-only set once a lookup of the viewer's own succeeds, not when it starts: an exact lookup marked `viewerLookup` (a name search, a label lookup, a mounted -link, a search read, or any `session.channels.resolve` call, which only the -viewer's own surfaces make) reports through `resolved`, and so does a mounted link +link, a search read, or a `session.channels.resolve` call with its default viewer lookup) reports through `resolved`, and so does a mounted link to a channel already known, which needs no lookup: links hold their demand even when readable. Rechecks, such as preview revalidation after an access-revoked CLOSED or a member-added confirmation, -leave it as it was. +leave it as it was. Watch admission calls `session.channels.resolve` with +`viewerLookup: false`, preserving agent-only provenance instead of promoting the channel into viewer mention traffic. Shared composers and reactions remain unavailable for nonmembers; message helpers and workflows do not acquire write eligibility from public metadata. The relay remains the final write authority, including low-level outbox intent. diff --git a/src/bundled/github/watcher.test.ts b/src/bundled/github/watcher.test.ts index 635699df73..ea43fd42f5 100644 --- a/src/bundled/github/watcher.test.ts +++ b/src/bundled/github/watcher.test.ts @@ -504,6 +504,33 @@ describe("github_watch_pr", () => { app.dispose(); }); + it("keeps the old Interest watch if Attention turns off during a move", async () => { + const app = setup(); + await app.call({ url }); + const id = Object.keys(app.records[agent] ?? {}).find((slug) => + slug.startsWith("watch/"), + ); + if (!id) throw new Error("Missing watch"); + const watch = app.records[agent]?.[id]; + if (watch?.type !== "source") throw new Error("Missing source"); + await app.save(agent, { + attention: { + "interest/github-pulls": { type: "interest", instructions: "Follow" }, + [id]: { ...watch, interest_id: "github-pulls" }, + }, + }); + const before = app.records[agent]?.[id]; + const read = app.states.read.bind(app.states); + vi.spyOn(app.states, "read").mockImplementationOnce(async (...args) => { + const result = await read(...args); + app.attentionEnabled(false); + return result; + }); + await expect(app.call({ url }, true)).rejects.toThrow(/Turn on Attention/); + expect(app.records[agent]?.[id]).toEqual(before); + app.dispose(); + }); + it("drops watches saved before Attention instead of importing them", async () => { localStorage.setItem( "buzz.github.pull-watches.v1", diff --git a/src/bundled/github/watcher.ts b/src/bundled/github/watcher.ts index 7c916b7e51..b8e9fc9708 100644 --- a/src/bundled/github/watcher.ts +++ b/src/bundled/github/watcher.ts @@ -512,33 +512,37 @@ export function watchPulls( // github-pulls Interest this plugin used to make, moves to this one. const owner = object && object.type !== "interest" ? object.interest_id : undefined; - let expected = priorObject.expected_state; - if (object && owner !== interest) - expected = ( - JSON.parse( - await callAttentionTool(attention, "watch_remove", { - id, - interest: owner, - expected_state: expected, - }), - ) as { expected_state: string } - ).expected_state; - const update = object && owner === interest; - await callAttentionTool( - attention, - update ? "watch_update" : "watch_add", - { - id, - interest, - expected_state: expected, - ...(update - ? fields.root - ? {} - : { clear: ["root"] } - : { type: "source" }), - ...fields, - }, - ); + if (object && object.type !== "interest" && owner !== interest) { + // Admission and the transfer share one write; a refusal keeps the old watch. + if (!attention.enabled()) + throw new Error("Turn on Attention to watch pull requests"); + await attention.write( + `watch/${id}`, + { type: "source", interest_id: interest, ...fields }, + { + expected: priorObject.expected_state, + interest: object.interest_id, + move: true, + }, + ); + } else { + const update = !!object; + await callAttentionTool( + attention, + update ? "watch_update" : "watch_add", + { + id, + interest, + expected_state: priorObject.expected_state, + ...(update + ? fields.root + ? {} + : { clear: ["root"] } + : { type: "source" }), + ...fields, + }, + ); + } return [ `Watching ${pull.url}. You will be woken here${moved ? ", no longer in the conversation that asked before," : ""} when CI finishes on its latest commit or new feedback arrives.`, ...(object && owner !== interest diff --git a/src/buzz-mcp/attention.ts b/src/buzz-mcp/attention.ts index 1c7325bc67..2ec2db3a8a 100644 --- a/src/buzz-mcp/attention.ts +++ b/src/buzz-mcp/attention.ts @@ -176,7 +176,7 @@ export const ATTENTION_TOOLS: readonly Tool[] = [ "watch_add", `Add an event watch (since and channels), a timer (prompt and interval_secs), a source subscription (provider and config) to an Interest; only for a new watch. -Event watch: it hears the channels you may read: those you are a member of, and public channels you have not joined. "all" means all of those, but hears a public channel you have not joined only about 30 seconds late. A delivery channel must be one you are a member of; the relay decides where you may post. It sees events that arrive while it is on: an earlier since does not replay older ones. Each matching event watch runs independently. Prefer deterministic selectors for hard rules; use the classifier only for judgment. authors keeps only those authors, exclude_authors: ["OWNER_HEX"] drops the owner's own messages, and is_reply: false keeps only top-level messages (true: only thread replies). A watch has authors or exclude_authors, not both. When a watch reports to the owner, exclude the owner's messages unless they are the signal. +Event watch: it hears the channels you may read: those you are a member of, and public channels you have not joined. "all" means all of those, but hears a public channel you have not joined only about 30 seconds late. A delivery channel must be one you are a member of; the relay decides where you may post. Enabling or editing an enabled watch refuses unreadable listening channels and delivery channels where membership is absent or unconfirmed. It sees events that arrive while it is on: an earlier since does not replay older ones. Each matching event watch runs independently. Prefer deterministic selectors for hard rules; use the classifier only for judgment. authors keeps only those authors, exclude_authors: ["OWNER_HEX"] drops the owner's own messages, and is_reply: false keeps only top-level messages (true: only thread replies). A watch has authors or exclude_authors, not both. When a watch reports to the owner, exclude the owner's messages unless they are the signal. Classifier, for noisy event watches or source subscriptions: Jev tests relevance after the normal filters. It sees one event or emitted source output, not history, and has no tools. Use named yes/no questions, for example {"questions":{"release_blocked":{"question":"Does the event say the release cannot proceed until something is fixed?","true":"It reports a release blocker, in any form: a question, an instruction or a status report.","false":"Routine release progress, or a problem that does not stop the release."}}}. Write questions for the facts the Interest needs, not for message types, and accept any phrasing of those facts. Ask one thing per question. The event passes when any question says yes, and each added question can add extra passes, so keep them few. For "A but not B", write one question that says so. Optional per question: true and false descriptions, guidance for examples and exclusions, and threshold (default 0.2; lower passes more). Change a threshold only when eval cases support it. Keep actions in the Interest. Pass uncertain cases. Errors pass events through. Every answer says classifier: available or unavailable. Classifiers can be saved either way and run once one is available; until then a classified watch's events pass unchecked, and the turn they start says so in a block. diff --git a/src/features/agents/attention-objects.test.ts b/src/features/agents/attention-objects.test.ts index 5cee5d2693..94cb5326f5 100644 --- a/src/features/agents/attention-objects.test.ts +++ b/src/features/agents/attention-objects.test.ts @@ -137,6 +137,49 @@ describe("checked writes", () => { ); }); + it("lets an internal helper move a watch with its current owner's token", async () => { + const start = record(); + const expected = (await show(start, "watch/asks", now)).expected_state; + const next = watch({ interest_id: "other" }); + const options = { expected, interest: "triage", move: true }; + const moved = await checkedWrite(start, "watch/asks", next, options, now); + expect(moved.attention["watch/asks"]?.value).toEqual(next); + await expect( + checkedWrite( + start, + "watch/asks", + next, + { ...options, interest: "other" }, + now, + ), + ).rejects.toThrow(/^wrong-scope:/); + await expect( + checkedWrite(moved, "watch/asks", next, options, now), + ).rejects.toThrow(/^wrong-scope:/); + const edited = await write(start, "watch/asks", watch({ name: "Edited" })); + await expect( + checkedWrite(edited, "watch/asks", next, options, now), + ).rejects.toThrow(/^conflict:/); + await expect( + checkedWrite( + start, + "watch/asks", + watch({ interest_id: "missing" }), + options, + now, + ), + ).rejects.toThrow(/^not-found:/); + await expect( + checkedWrite( + start, + "watch/asks", + timer({ interest_id: "other" }), + options, + now, + ), + ).rejects.toThrow(/^invalid-operation:.*type/); + }); + it("remove an Interest only after its watches", async () => { const start = record(); await expect(write(start, "interest/triage", null)).rejects.toThrow( diff --git a/src/features/agents/attention-objects.ts b/src/features/agents/attention-objects.ts index b5b4434b1e..44cb616e09 100644 --- a/src/features/agents/attention-objects.ts +++ b/src/features/agents/attention-objects.ts @@ -269,6 +269,8 @@ export type WriteOptions = Readonly<{ /** A timer rearm: its deadline is set from `armed_at` even when that is * unchanged, keeping the occurrences used. */ rearm?: boolean; + /** Internal helpers may transfer a watch atomically, scoped to its current Interest. */ + move?: boolean; }>; /** The record after one checked write: replaces (or with `null` removes) the @@ -277,7 +279,7 @@ export async function checkedWrite( record: AgentRecord, slug: string, value: AttentionValue | null, - { expected, interest, rearm }: WriteOptions, + { expected, interest, rearm, move }: WriteOptions, now: number, ): Promise { const parsed = parseSlug(slug); @@ -337,7 +339,8 @@ export async function checkedWrite( value && prior.type !== "interest" && value.type !== "interest" && - prior.interest_id !== value.interest_id + prior.interest_id !== value.interest_id && + !move ) throw new AttentionError( "invalid-operation", diff --git a/src/features/agents/runner.ts b/src/features/agents/runner.ts index 69ee97be1e..a30c452efc 100644 --- a/src/features/agents/runner.ts +++ b/src/features/agents/runner.ts @@ -57,6 +57,7 @@ import { } from "./attention"; import { checkedWrite, + AttentionError, interestIds, show, watchSummaries, @@ -431,6 +432,12 @@ export type RunnerHost = { * channel is public; undefined as for `member`. Only a member may be * delivered to. */ reads(agent: Agent, event: Pick): boolean | undefined; + /** Resolves listening metadata before an enabled watch is admitted. */ + resolveChannels?( + agent: Agent, + channels: readonly string[], + signal: AbortSignal, + ): Promise; /** Whether the event's channel is a DM between the agent and one other. */ dm(agent: Agent, event: RelayEvent): boolean; /** Whether the event came as it would have with no watch keeping channels @@ -3297,6 +3304,36 @@ export class AgentRunners { ); lifetime.throwIfAborted(); check?.(); + if (value && value.type !== "interest" && value.enabled) { + const agent = this.host.agent(key); + if (!agent) throw new Error("This agent is no longer running"); + if (value.type === "event" && value.channels !== "all") { + await this.host.resolveChannels?.( + agent, + value.channels, + lifetime, + ); + lifetime.throwIfAborted(); + check?.(); + for (const channel of value.channels) + if ( + this.host.reads(agent, { tags: [["h", channel]] }) !== true + ) + throw new AttentionError( + "invalid-operation", + `Cannot receive events from channel ${channel}. Check channel access and wait for confirmed metadata before enabling this watch.`, + ); + } + if ( + value.channel && + this.host.member(agent, { tags: [["h", value.channel]] }) !== + true + ) + throw new AttentionError( + "invalid-operation", + `Agent is not a member of the delivery channel ${value.channel}, or membership is unconfirmed. Join it and wait for confirmation before enabling this watch.`, + ); + } return next; }); return show(after, slug, seconds()); diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index cf25c2350d..f022639921 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -1,3 +1,7 @@ +import { watchPulls } from "../../bundled/github/watcher"; +import type { PullStates } from "../../bundled/github/pull-states"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import type { BuzzClient } from "../../buzz-mcp/client"; import { federatedFixture } from "../federated-identity/testing"; import { workingAgentDetails } from "../../bundled/channels/working-agents"; import { completionTarget } from "../notifications/agent-completions"; @@ -34,6 +38,7 @@ import { bytesToHex } from "nostr-tools/utils"; import { finalizeEvent, generateSecretKey } from "nostr-tools"; import { afterEach, + onTestFinished, beforeAll, beforeEach, describe, @@ -11626,6 +11631,334 @@ it("lets a run read and write its attention with expected-state tokens", async ( ).rejects.toThrow(/^conflict:/); }); +/** Valid tool channel ids for nonmember, cached and unknown metadata. */ +const refusedChannels = [ + "10000000-0000-4000-8000-000000000001", + "10000000-0000-4000-8000-000000000002", + "10000000-0000-4000-8000-000000000003", +] as const; +function watchChannelMetadata(session: RelaySession) { + const snapshot = session.channels.list(); + vi.spyOn(session.channels, "list").mockReturnValue({ + ...snapshot, + channels: snapshot.channels.map((row) => ({ + ...row, + id: + row.id === "private" + ? refusedChannels[0] + : row.id === "saved" + ? refusedChannels[1] + : row.id, + })), + }); +} + +it("refuses watch channels delivery cannot receive, without saving add or update", async () => { + const { service, run, emit, session } = await setup(); + watchChannelMetadata(session); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + for (const target of refusedChannels) { + await expect( + callAttentionTool(attention, "watch_add", { + id: "unreachable", + interest: "default", + expected_state: "object-v1:absent", + type: "event", + channels: [channel, target], + kinds: [9], + since: "now", + }), + ).rejects.toThrow(/cannot receive events.*channel/i); + expect((await attention.show("watch/unreachable")).object).toBeNull(); + } + await callAttentionTool(attention, "watch_add", { + id: "reachable", + interest: "default", + expected_state: "object-v1:absent", + type: "event", + channels: [channel], + kinds: [9], + since: "now", + }); + const before = await attention.show("watch/reachable"); + await expect( + callAttentionTool(attention, "watch_update", { + id: "reachable", + interest: "default", + expected_state: before.expected_state, + channels: [refusedChannels[0]], + since: "now", + }), + ).rejects.toThrow(/cannot receive events.*channel/i); + expect(await attention.show("watch/reachable")).toEqual(before); +}); + +it.each(["event", "timer", "source"] as const)( + "refuses an unreachable %s delivery channel", + async (type) => { + const { service, run, emit, session } = await setup(); + watchChannelMetadata(session); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + await expect( + callAttentionTool(attention, "watch_add", { + id: "unreachable", + interest: "default", + expected_state: "object-v1:absent", + type, + channel: refusedChannels[0], + ...(type === "event" + ? { channels: [channel], since: "now" } + : type === "timer" + ? { interval_secs: 60, prompt: "Check" } + : { provider: "attention/command", config: { command: ["true"] } }), + }), + ).rejects.toThrow(/not a member.*delivery channel/i); + expect((await attention.show("watch/unreachable")).object).toBeNull(); + }, +); + +it.each(refusedChannels)( + "preserves a GitHub watch when a move to %s is refused", + async (target) => { + const { service, run, emit, session } = await setup(); + watchChannelMetadata(session); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + vi.stubGlobal("localStorage", { + getItem: () => null, + setItem: () => {}, + removeItem: () => {}, + }); + onTestFinished(() => { + vi.unstubAllGlobals(); + }); + const tools: AddedTool[] = []; + watchPulls( + { + agents: { + addAttentionSource: () => {}, + addTool: (tool: AddedTool) => tools.push(tool), + }, + effect: () => {}, + } as unknown as Context, + { + read: async () => ({ + state: { + state: "open", + head: "1111111111", + feedback: { viewer: "alice", ids: [], items: [] }, + }, + }), + } as unknown as PullStates, + undefined, + ); + const tool = tools[0]; + if (!tool) throw new Error("Missing GitHub watch tool"); + const client = { key: bot, pubkey: bot, attention } as BuzzClient; + const url = "https://github.com/block/buzz-app/pull/2311"; + await tool.call( + client, + { channel, topLevel: true }, + { url, interest: "default" }, + ); + const watch = attention.watches({ interest: "default", type: "source" })[0]; + if (!watch) throw new Error("Missing GitHub watch"); + const slug = `watch/${watch.id}`; + const before = await attention.show(slug); + await attention.write( + "interest/moved", + { type: "interest", instructions: "Follow the PR here" }, + { expected: "object-v1:absent" }, + ); + await expect( + tool.call( + client, + { channel: target, topLevel: true }, + { url, interest: "moved" }, + ), + ).rejects.toThrow(/not a member.*delivery channel/i); + expect(await attention.show(slug)).toEqual(before); + // Retry after returning to a confirmed destination, using the same watch. + await tool.call( + client, + { channel, topLevel: true }, + { url, interest: "moved" }, + ); + expect((await attention.show(slug)).object).toMatchObject({ + interest_id: "moved", + channel, + }); + expect(attention.watches({ type: "source" })).toHaveLength(1); + }, +); + +it("resolves an unopened public listening channel before admission and delivers its events", async () => { + const { service, run, emit, session } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + const target = refusedChannels[2]; + let resolved = false; + const get = session.channels.get?.bind(session.channels); + vi.spyOn(session.channels, "get").mockImplementation((id) => + id === target + ? resolved + ? { id, name: id, visibility: "public", members: [other] } + : undefined + : get?.(id), + ); + const resolve = vi.fn(async (ids: readonly string[]) => { + expect(ids).toEqual([target]); + resolved = true; + }); + Object.assign(session.channels, { resolve }); + await callAttentionTool(attention, "watch_add", { + id: "unopened", + interest: "default", + expected_state: "object-v1:absent", + type: "event", + channels: [target], + kinds: [9], + since: 1, + }); + expect(resolve).toHaveBeenCalledWith( + [target], + expect.objectContaining({ viewerLookup: false }), + ); + const called = run.mock.calls.length; + emit({ events: [event("unopened-event", { tags: [["h", target]] })] }); + await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(called + 1)); + expect(run.mock.calls.at(-1)?.[0].trigger).toMatchObject({ + type: "watch", + slug: "watch/unopened", + }); + await expect( + callAttentionTool(attention, "watch_update", { + id: "unopened", + interest: "default", + expected_state: (await attention.show("watch/unopened")).expected_state, + channel: target, + }), + ).rejects.toThrow(/not a member.*delivery channel/i); +}); + +it.each(["private", "empty", "failed"] as const)( + "preserves a watch when listening metadata resolves %s", + async (result) => { + const { service, run, emit, session } = await setup(); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + await callAttentionTool(attention, "watch_add", { + id: "kept", + interest: "default", + expected_state: "object-v1:absent", + type: "event", + channels: [channel], + kinds: [9], + since: 1, + }); + const before = await attention.show("watch/kept"); + const target = refusedChannels[2]; + let resolved = false; + const get = session.channels.get?.bind(session.channels); + vi.spyOn(session.channels, "get").mockImplementation((id) => + id === target && resolved && result === "private" + ? { id, name: id, visibility: "private", members: [other] } + : get?.(id), + ); + const resolve = vi.fn(async () => { + if (result === "failed") throw new Error("Metadata read failed"); + resolved = true; + }); + Object.assign(session.channels, { resolve }); + await expect( + callAttentionTool(attention, "watch_update", { + id: "kept", + interest: "default", + expected_state: before.expected_state, + channels: [target], + }), + ).rejects.toThrow( + result === "failed" + ? /Metadata read failed/ + : /cannot receive events.*channel/i, + ); + expect(resolve).toHaveBeenCalledOnce(); + expect(await attention.show("watch/kept")).toEqual(before); + }, +); + +it("uses one listening policy for watch admission and event delivery", async () => { + const { service, run, emit, session } = await setup(); + watchChannelMetadata(session); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + // Listening access may be broader than destination membership; admission and delivery share the policy. + vi.spyOn( + service as unknown as { + reads(agent: Agent, event: Pick): boolean; + }, + "reads", + ).mockReturnValue(true); + await callAttentionTool(attention, "watch_add", { + id: "policy", + interest: "default", + expected_state: "object-v1:absent", + type: "event", + channels: [refusedChannels[0]], + kinds: [9], + since: 1, + }); + const called = run.mock.calls.length; + emit({ + events: [event("policy-event", { tags: [["h", refusedChannels[0]]] })], + }); + await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(called + 1)); + expect(run.mock.calls.at(-1)?.[0].trigger).toMatchObject({ + type: "watch", + slug: "watch/policy", + }); +}); + +it("can disable an unreachable watch, but cannot enable it until channel membership is confirmed", async () => { + const { service, run, emit, session } = await setup(); + watchChannelMetadata(session); + await service.create({ type: "example/echo", name: "Echo" }); + await service.save(bot, { attentionEnabled: true }); + const { attention } = await handleOf(run, emit); + await callAttentionTool(attention, "watch_add", { + id: "disabled", + interest: "default", + expected_state: "object-v1:absent", + type: "event", + enabled: false, + channels: [refusedChannels[0]], + kinds: [9], + since: "now", + }); + const before = await attention.show("watch/disabled"); + await expect( + callAttentionTool(attention, "watch_enable", { + id: "disabled", + interest: "default", + expected_state: before.expected_state, + }), + ).rejects.toThrow(/cannot receive events.*channel/i); + await callAttentionTool(attention, "watch_remove", { + id: "disabled", + interest: "default", + expected_state: before.expected_state, + }); + expect((await attention.show("watch/disabled")).object).toBeNull(); +}); + it("fails an agent's attention write that agent storage could not save, and keeps the saved value", async () => { let failing = false; const storage = fileStore(new Map(), async () => { diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index 394e56234d..0032c6a33b 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -2128,6 +2128,14 @@ export class AgentsService extends Service implements Agents { agent: (key) => this.agent(key), member: (agent, event) => this.member(agent, event), reads: (agent, event) => this.reads(agent, event), + resolveChannels: async (agent, ids, signal) => { + const channels = this.bindingOf(agent)?.session.channels; + const unknown = ids.filter( + (id) => !this.channelOf(agent, { tags: [["h", id]] }), + ); + if (unknown.length) + await channels?.resolve?.(unknown, { signal, viewerLookup: false }); + }, dm: (agent, event) => this.dm(agent, event), unattended: (agent, event, live) => { const channel = event.tags.find((tag) => tag[0] === "h")?.[1]; diff --git a/src/features/relay/attend.session.test.ts b/src/features/relay/attend.session.test.ts index 3e5886329f..e019d3651a 100644 --- a/src/features/relay/attend.session.test.ts +++ b/src/features/relay/attend.session.test.ts @@ -946,54 +946,62 @@ it("counts a joined channel's metadata as only saved from a stream drop until di expect(summary()?.metadataCached).toBeUndefined(); }); -it("keeps a channel the viewer looked up heard for the viewer when a recheck after an access-revoked CLOSED finds it public again", async () => { - const h = harness(); - await h.discover(); - const lookup = () => - h.wire.pending.findIndex( - (entry) => !entry.signal?.aborted && asks(entry.filters, OPEN), - ); - // The viewer's own lookup, for a link since unmounted, finds it before - // agents attend it. - const release = h.owner.session.channels.refer?.(OPEN); - await vi.waitFor(() => expect(lookup()).not.toBe(-1)); - h.wire.pending - .splice(lookup(), 1)[0] - ?.respond(channel(OPEN, "public") as RelayEvent[]); - await vi.waitFor(() => - expect(h.owner.session.channels.get?.(OPEN)).toBeDefined(), - ); - release?.(); - const stop = h.owner.session.attend([OPEN]); - await vi.waitFor(() => - expect(h.update.mock.lastCall?.[0] ?? []).toContain(OPEN), - ); - expect(h.owner.session.unattended(OPEN, false)).toBe(true); - h.state({ - status: "connected", - routes: [ - { - id: "attended", - channelId: OPEN, - status: "error", - replay: "unknown", - error: "restricted: channel access revoked", - }, - ], - }); - // The recheck, and attendance's own read of the suspended channel, find it - // public again; neither is news to the viewer. - await vi.waitFor(() => expect(lookup()).not.toBe(-1)); - await flush(); - while (lookup() !== -1) +it.each([false, true])( + "keeps a viewer preview heard after access recovery with admission lookup %s", + async (admit) => { + const h = harness(); + await h.discover(); + const lookup = () => + h.wire.pending.findIndex( + (entry) => !entry.signal?.aborted && asks(entry.filters, OPEN), + ); + // The viewer's own lookup, for a link since unmounted, finds it before + // agents attend it. + const release = h.owner.session.channels.refer?.(OPEN); + await vi.waitFor(() => expect(lookup()).not.toBe(-1)); h.wire.pending .splice(lookup(), 1)[0] ?.respond(channel(OPEN, "public") as RelayEvent[]); - await flush(); - expect(h.owner.session.channels.get?.(OPEN)).toBeDefined(); - expect(h.owner.session.unattended(OPEN, false)).toBe(true); - stop(); -}); + await vi.waitFor(() => + expect(h.owner.session.channels.get?.(OPEN)).toBeDefined(), + ); + release?.(); + const stop = h.owner.session.attend([OPEN]); + await vi.waitFor(() => + expect(h.update.mock.lastCall?.[0] ?? []).toContain(OPEN), + ); + expect(h.owner.session.unattended(OPEN, false)).toBe(true); + h.state({ + status: "connected", + routes: [ + { + id: "attended", + channelId: OPEN, + status: "error", + replay: "unknown", + error: "restricted: channel access revoked", + }, + ], + }); + // Admission may resolve the temporarily hidden preview before recovery. + expect(h.owner.session.channels.get?.(OPEN)).toBeUndefined(); + const admission = admit + ? h.owner.session.channels.resolve?.([OPEN], { viewerLookup: false }) + : undefined; + // The recheck, attendance and admission preserve the viewer's provenance. + await vi.waitFor(() => expect(lookup()).not.toBe(-1)); + await flush(); + while (lookup() !== -1) + h.wire.pending + .splice(lookup(), 1)[0] + ?.respond(channel(OPEN, "public") as RelayEvent[]); + await flush(); + await admission; + expect(h.owner.session.channels.get?.(OPEN)).toBeDefined(); + expect(h.owner.session.unattended(OPEN, false)).toBe(true); + stop(); + }, +); it("skips the queued lookups of attended channels no watch wants any more, so a new watch's lookup does not wait behind them", async () => { const h = harness(); @@ -1337,6 +1345,55 @@ it("keeps a suspended preview the viewer looked up heard for the viewer when a r expect(h.owner.session.unattended(OPEN, false)).toBe(true); stop(); }); +it("keeps explicit agent metadata admission out of viewer provenance", async () => { + const h = harness(); + await h.discover(); + const resolving = h.owner.session.channels.resolve?.([OPEN], { + viewerLookup: false, + }); + assert(resolving); + await vi.waitFor(() => + expect(h.wire.pending.some((entry) => asks(entry.filters, OPEN))).toBe( + true, + ), + ); + h.wire.pending + .splice( + h.wire.pending.findIndex((entry) => asks(entry.filters, OPEN)), + 1, + )[0] + ?.respond(channel(OPEN, "public") as RelayEvent[]); + await resolving; + h.owner.session.attend([OPEN]); + expect(h.owner.session.unattended(OPEN, false)).toBe(false); +}); + +it.each(["loading", "error"] as const)( + "refuses agent metadata admission while discovery is %s", + async (status) => { + const h = harness(); + h.connect(); + await vi.waitFor(() => expect(h.wire.pending).toHaveLength(1)); + if (status === "error") h.wire.next().fail(new Error("Discovery failed")); + await vi.waitFor(() => + expect(h.owner.session.channels.list().status).toBe(status), + ); + const before = h.owner.session.channels.list(); + const resolving = h.owner.session.channels.resolve?.([OPEN], { + viewerLookup: false, + }); + assert(resolving); + const outcome = resolving.then( + () => "resolved", + (error: Error) => error.message, + ); + await flush(); + const exact = h.wire.pending.find((entry) => asks(entry.filters, OPEN)); + exact?.respond(channel(OPEN, "public") as RelayEvent[]); + expect(await outcome).toMatch(/not discovered/i); + expect(h.owner.session.channels.list()).toEqual(before); + }, +); it("stops counting an attended preview's saved grant once a read after a drop omits it, and hears it once a later lookup finds it public again", async () => { for (const continuous of [false, true]) { diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index 0074781ea5..97bf4ac6ac 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -1210,13 +1210,21 @@ export function createRelaySession( const projectedChannels = new WeakMap(); const channelQueries = Object.freeze({ ...channels.queries, - // Only the viewer's own surfaces call it from outside the session (a - // scoped search, a requested channel, a repository page), so what they - // look up becomes the viewer's, as without any `attend`. + // Viewer surfaces promote their lookups by default. Explicit agent lookups + // extend discovery without turning watch-only channels into mention targets. ...(channels.queries.resolve && { - resolve: (ids: readonly string[], options?: ChannelReadOptions) => - channels.queries.resolve?.(ids, { ...options, viewerLookup: true }) ?? - Promise.resolve(), + resolve: async (ids: readonly string[], options?: ChannelReadOptions) => { + const viewerLookup = options?.viewerLookup !== false; + if (!viewerLookup) { + if (channels.queries.list().status !== "ready") + throw new ReadError( + "unavailable", + "Channels are not discovered yet.", + ); + for (const id of ids) learnedByAgents(id); + } + await channels.queries.resolve?.(ids, { ...options, viewerLookup }); + }, }), list() { const snapshot = channels.queries.list();