feat(service): add durable autonomous conversation participation V2

docs/adr/ADR-0001-discord-ambient-worker-topology.md

@@ -18,7 +18,7 @@ Each allowed guild/channel gets its own lease key, so one worker process can own
 
 Guild/user relationships are bounded and idempotent. Membership v1 uses complete paged guild rosters; active humans are recent (10-minute) non-bot authors intersected with a fresh complete roster. First polling seeds a cursor without replying; later work is durable. Raw events are marked archived after 14 days, while raw events and source-linked summaries remain permanently: neither is deleted.
 
-Ambient participation is continuous and probabilistic. A normal request for silence is social transcript evidence: the model may accept, ignore, resist, or escalate. It must never become deterministic mute, quit, or quiet-until state. Only operational kill switches, lease loss, disabled mappings, and invalid startup configuration are deterministic. Delivery uses one to five typed bubbles, bounded length delay, durable nonces/receipts, and cancels remaining bubbles on newer human ingress.
+Ambient participation is continuous and probabilistic. A normal request for silence is social transcript evidence: the model may accept, ignore, resist, or escalate. It must never become deterministic mute, quit, or quiet-until state. Only operational kill switches, lease loss, disabled mappings, and invalid startup configuration are deterministic. Active human conversation is sufficient evidence for spontaneous participation; an explicit mention or direct address is not required. Each poll captures the newest actionable event as a fixed high-watermark, appraises the recent conversation once, and atomically marks older queued work through that boundary as observed. Events ingested after the boundary remain pending for the next poll and do not cancel delivery from the current poll. Delivery uses one to five typed bubbles, bounded length delay, and durable nonces/receipts.
 
 ## Ambient hardening addendum
 

docs/agent-runbook.md

@@ -133,7 +133,7 @@ Set per-channel overrides in `channel_settings.settings_json`; absent or invalid
 | Key | Default |
 | --- | --- |
 | `ambientBudgetPerHour` | `20` |
-| `ambientConfidenceFloor` | `0.7` |
+| `ambientConfidenceFloor` | `0.6` |
 | `ambientIdleDecayTauMs` | `7200000` (2h) |
 | `ambientPressureTauMs` | `1800000` (30m) |
 | `ambientPityEnabled` | `true` |

service/src/adaptive-ambient-provider.test.ts

@@ -11,7 +11,7 @@ type ConversationProviderClient = { readonly complete: (prompt: ConversationProm
 type AmbientAppraisalResult =
   | { readonly kind: "valid"; readonly proposal: { readonly decision: "observe" | "speak"; readonly chunks: readonly string[] }; readonly diagnostic?: string }
   | { readonly kind: "invalid"; readonly diagnostic: string };
-type AppraisalAudience = { readonly rosterComplete: boolean; readonly activeHumanCount: number; readonly currentDrive: number; readonly budgetRemaining: number };
+type AppraisalAudience = { readonly rosterComplete: boolean; readonly currentDrive: number; readonly budgetRemaining: number };
 type AdaptiveAmbientProvider = { readonly appraise: (request: { readonly scope: { readonly guildId: string; readonly channelId: string }; readonly persona: string; readonly transcript: readonly DiscordInboundMessage[]; readonly audience?: AppraisalAudience }, options?: { readonly signal?: AbortSignal }) => Promise<AmbientAppraisalResult> };
 type AdaptiveAmbientProviderApi = {
   readonly createOpenAiConversationProviderClient: (config: { readonly endpoint: URL | string; readonly token: string; readonly model: string; readonly timeoutMs: number; readonly fetchImpl?: typeof fetch }) => ConversationProviderClient;
@@ -87,6 +87,12 @@ describe("strict adaptive ambient appraisal provider", () => {
     expect(system).toContain("Never claim human identity");
     expect(system).toContain("Silence in the room is never a reason to speak.");
     expect(system).toContain("Never answer a question addressed to another participant; only respond when the conversational context invites you.");
+    expect(system).toContain("one conversation batch");
+    expect(system).toContain("choose speak by default");
+    expect(system).toContain("explicit mention");
+    expect(system).toContain("may initiate");
+    expect(system).toContain("participationPrior");
+    expect(system).toContain("do not treat observe as the default");
     expect(system).not.toContain("ambient-provider-test-secret");
     expect(JSON.parse(user)).toMatchObject({ transcript });
     expect(JSON.parse(user)).not.toHaveProperty("audience");
@@ -95,15 +101,15 @@ describe("strict adaptive ambient appraisal provider", () => {
 
   it("includes injected audience context in the appraisal payload", async () => {
     let wireBody: { messages: Array<{ role: string; content: string }> } | undefined;
-    const audience = { rosterComplete: true, activeHumanCount: 3, currentDrive: 0.7, budgetRemaining: 4 };
+    const audience = { rosterComplete: true, currentDrive: 0.7, budgetRemaining: 4 };
     const fetchImpl = vi.fn(async (_: URL | RequestInfo, init?: RequestInit) => {
       wireBody = JSON.parse(String(init?.body));
       return chatResponse(validAppraisal());
     }) as typeof fetch;
 
     await ambientProvider(fetchImpl).appraise(request(audience));
 
-    expect(JSON.parse(wireBody?.messages[1]?.content ?? "{}")).toMatchObject({ audience });
+    expect(JSON.parse(wireBody?.messages[1]?.content ?? "{}")).toMatchObject({ audience, participationPrior: { speak: 0.9, observe: 0.1 } });
   });
 
   it("turns every provider and contract failure into invalid audit input without leaking secrets", async () => {

service/src/adaptive-ambient-provider.ts

@@ -15,7 +15,7 @@ export type AmbientAppraisalRequest = {
     readonly archiveSummaries: readonly string[];
     readonly relationships: readonly { readonly userId: string; readonly rapport: number; readonly familiarity: number; readonly notes: readonly string[] }[];
   };
-  readonly audience?: { readonly rosterComplete: boolean; readonly activeHumanCount: number; readonly currentDrive: number; readonly budgetRemaining: number };
+  readonly audience?: { readonly rosterComplete: boolean; readonly currentDrive: number; readonly budgetRemaining: number };
 };
 
 export type AdaptiveAmbientAppraisalResult = AmbientAppraisalParseResult & { readonly diagnostic?: string };
@@ -70,6 +70,10 @@ function buildAdaptiveAmbientAppraisalPrompt(request: AmbientAppraisalRequest):
       "Never claim human identity.",
       "Silence in the room is never a reason to speak.",
       "Never answer a question addressed to another participant; only respond when the conversational context invites you.",
+      "Treat the transcript as one conversation batch and choose one timely contribution to the overall exchange, not a reply to every message.",
+      "For every non-stale human conversation batch, choose speak by default; choose observe only when every possible contribution would be irrelevant, repetitive, intrusive, or directed at another participant.",
+      "An explicit mention, direct address, or reply is not required: the persona may initiate a reaction, observation, joke, question, or topic shift from the active conversation.",
+      "Use participationPrior as the starting decision prior before considering transcript evidence; do not treat observe as the default class.",
       "Required fields: schema, decision, desiredDrive, confidence, chunks, relationshipProposals.",
       "desiredDrive and confidence must be JSON numbers between 0 and 1, never strings, words, or percentages.",
       "decision is observe or speak; observe requires chunks []; speak requires one to five non-empty chunks no longer than 1800 characters.",
@@ -81,11 +85,15 @@ function buildAdaptiveAmbientAppraisalPrompt(request: AmbientAppraisalRequest):
       persona: request.persona,
       transcript: request.transcript,
       context: request.context ?? { archiveSummaries: [], relationships: [] },
-      ...(request.audience === undefined ? {} : { audience: request.audience }),
+      ...(request.audience === undefined ? {} : { audience: request.audience, participationPrior: participationPrior() }),
     }),
   };
 }
 
+function participationPrior(): { readonly speak: number; readonly observe: number } {
+  return { speak: 0.9, observe: 0.1 };
+}
+
 function completionOptions(model: string | undefined, signal: AbortSignal | undefined) {
   return { ...(model ? { model } : {}), signal };
 }

service/src/adaptive-ambient-runtime.test.ts

@@ -82,8 +82,8 @@ describe("atomic adaptive ambient runtime", () => {
     await expect(planned.runtime!.run({ fence: planned.fence, signal: new AbortController().signal })).resolves.toBe("planned");
     const plannedAudit = auditEvidence(planned.db);
     expect(plannedAudit).toEqual({
-      evidenceWeight: 1, probability: 0.625, draw: service.stableAmbientDraw(`${scope.guildId}:${scope.channelId}`, "event-1"),
-      driveBefore: 0.5, driveAfter: 0.625, activeHumanCount: 1, rosterFresh: 1,
+      evidenceWeight: 1, probability: 1, draw: service.stableAmbientDraw(`${scope.guildId}:${scope.channelId}`, "event-1"),
+      driveBefore: 0.7, driveAfter: 0.7 * 0.75 + 1 * 0.25, activeHumanCount: 1, rosterFresh: 1,
     });
     if (plannedAudit.driveBefore === null || plannedAudit.driveAfter === null) throw new Error("planned audit must retain drive evidence");
     expect(service.calculateNextAmbientDrive({ scope, drive: plannedAudit.driveBefore, version: 0, updatedAtMs: now }, 1)).toBeCloseTo(plannedAudit.driveAfter, 12);
@@ -97,7 +97,7 @@ describe("atomic adaptive ambient runtime", () => {
     await expect(observed.runtime!.run({ fence: observed.fence, signal: new AbortController().signal })).resolves.toBe("observe");
     expect(auditEvidence(observed.db)).toEqual({
       evidenceWeight: 1, probability: 0, draw: service.stableAmbientDraw(`${scope.guildId}:${scope.channelId}`, "event-1"),
-      driveBefore: 0.5, driveAfter: 0.575, activeHumanCount: 1, rosterFresh: 1,
+      driveBefore: 0.7, driveAfter: 0.7 * 0.75 + 0.8 * 0.25, activeHumanCount: 1, rosterFresh: 1,
     });
     observed.db.close();
 
@@ -166,14 +166,14 @@ describe("atomic adaptive ambient runtime", () => {
     } }, startup: { enabled: true, allowlist: [scope], diagnostics: [] }, scope, botUserId, budgetPerHour: 2, clock: () => now, loadRoster: async () => roster() });
 
     await expect(fixture.runtime!.run({ fence: fixture.fence, signal: new AbortController().signal })).resolves.toBe("planned");
-    expect(audience).toEqual({ rosterComplete: true, activeHumanCount: 1, currentDrive: 0.8, budgetRemaining: 1 });
+    expect(audience).toEqual({ rosterComplete: true, currentDrive: 0.8, budgetRemaining: 1 });
     fixture.db.close();
   });
 
   it("persists unchanged streaks for a valid provider observe", async () => {
     const fixture = setup({ result: appraisal({ decision: "observe", desiredDrive: 0.8, chunks: [] }) });
     await expect(fixture.runtime!.run({ fence: fixture.fence, signal: new AbortController().signal })).resolves.toBe("observe");
-    expect(fixture.store.state(scope)).toEqual({ drive: 0.575, version: 1, updatedAtMs: now, pressure: 0, pressureUpdatedAtMs: now, speakStreak: 0, skipStreak: 0 });
+    expect(fixture.store.state(scope)).toEqual({ drive: 0.7 * 0.75 + 0.8 * 0.25, version: 1, updatedAtMs: now, pressure: 0, pressureUpdatedAtMs: now, speakStreak: 0, skipStreak: 0 });
     expect(fixture.store.counts()).toEqual({ audits: 1, states: 1, budgets: 0, relationships: 1, plans: 0 });
     expect(fixture.db.db.prepare("SELECT status FROM participant_event_work WHERE id='work-1'").get()).toEqual({ status: "observe" });
     fixture.db.close();
@@ -198,17 +198,17 @@ describe("atomic adaptive ambient runtime", () => {
     fixture.db.close();
   });
 
-  it("honors scoped decay, pressure, and disabled pity settings end-to-end", async () => {
+  it("honors scoped decay and pressure settings end-to-end", async () => {
     const eventId = Array.from({ length: 100 }, (_, index) => `settings-${index}`).find((candidate) => {
       const draw = service.stableAmbientDraw(`${scope.guildId}:${scope.channelId}`, candidate);
       return draw > 0.6 && draw < 0.9;
     });
     if (!eventId) throw new Error("could not select a deterministic ambient draw");
-    const fixture = setup({ eventId, settings: { ambientIdleDecayTauMs: 60_000, ambientPressureTauMs: 60_000, ambientPityEnabled: false }, result: appraisal({ desiredDrive: 0.1 }) });
+    const fixture = setup({ eventId, settings: { ambientIdleDecayTauMs: 60_000, ambientPressureTauMs: 60_000 }, result: appraisal({ desiredDrive: 0.1 }) });
     fixture.db.db.prepare(`INSERT INTO adaptive_ambient_state (guild_id,channel_id,drive,version,updated_at_ms,pressure,pressure_updated_at_ms,speak_streak,skip_streak)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(scope.guildId, scope.channelId, 1, 4, now - 60_000, 0.5, now - 60_000, 0, 4);
 
-    await expect(fixture.runtime!.run({ fence: fixture.fence, signal: new AbortController().signal })).resolves.toBe("observe");
+    await expect(fixture.runtime!.run({ fence: fixture.fence, signal: new AbortController().signal })).resolves.toBe("planned");
     const decayedDrive = 0.5 + 0.5 * Math.exp(-1);
     const decayedPressure = 0.5 * Math.exp(-1);
     expect(fixture.store.state(scope)).toEqual({
@@ -217,8 +217,8 @@ describe("atomic adaptive ambient runtime", () => {
       updatedAtMs: now,
       pressure: decayedPressure,
       pressureUpdatedAtMs: now,
-      speakStreak: 0,
-      skipStreak: 5,
+      speakStreak: 1,
+      skipStreak: 0,
     });
     fixture.db.close();
   });
@@ -243,8 +243,8 @@ describe("atomic adaptive ambient runtime", () => {
     fixture.db.close();
   });
 
-  it("accepts provider confidence at the scoped floor override", async () => {
-    const fixture = setup({ settings: { ambientConfidenceFloor: 0.6 }, result: appraisal({ confidence: 0.65 }) });
+  it("accepts provider confidence at the global default floor", async () => {
+    const fixture = setup({ result: appraisal({ confidence: 0.65 }) });
     await expect(fixture.runtime!.run({ fence: fixture.fence, signal: new AbortController().signal })).resolves.toBe("planned");
     expect(fixture.store.counts()).toEqual({ audits: 1, states: 1, budgets: 1, relationships: 1, plans: 1 });
     fixture.db.close();
@@ -262,7 +262,7 @@ describe("atomic adaptive ambient runtime", () => {
     const invalid = setup({ result: { kind: "invalid", diagnostic: "malformed provider result" } });
     await expect(invalid.runtime!.run({ fence: invalid.fence, signal: new AbortController().signal })).resolves.toBe("invalid");
     expect(invalid.store.counts()).toEqual({ audits: 1, states: 0, budgets: 0, relationships: 0, plans: 0 }); invalid.db.close();
-    const lowConfidence = setup({ result: appraisal({ confidence: 0.69 }) });
+    const lowConfidence = setup({ result: appraisal({ confidence: 0.59 }) });
     await expect(lowConfidence.runtime!.run({ fence: lowConfidence.fence, signal: new AbortController().signal })).resolves.toBe("invalid");
     expect(lowConfidence.store.counts()).toEqual({ audits: 1, states: 0, budgets: 0, relationships: 0, plans: 0 }); lowConfidence.db.close();
     const thrown = setup();

service/src/adaptive-ambient-runtime.ts

@@ -40,8 +40,9 @@ export type AdaptiveAmbientRuntime = {
 };
 
 const BUDGET_KEY = "ambient";
-const DEFAULT_AMBIENT_DRIVE = 0.5;
-const RECENT_CONTEXT_LIMIT = 20;
+const DEFAULT_AMBIENT_DRIVE = 0.7;
+const DEFAULT_AMBIENT_CONFIDENCE_FLOOR = 0.6;
+const RECENT_CONTEXT_LIMIT = 100;
 const ROSTER_FRESHNESS_MS = 5 * 60_000;
 
 export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOptions): AdaptiveAmbientRuntime {
@@ -87,8 +88,7 @@ export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOpti
           context: { archiveSummaries: context.archiveSummaries, relationships: context.relationships },
           audience: {
             rosterComplete: roster.complete,
-            activeHumanCount: activeHumanIds.length,
-            currentDrive: state?.drive ?? 0.5,
+            currentDrive: state?.drive ?? DEFAULT_AMBIENT_DRIVE,
             budgetRemaining: budgetRemaining(options.store, options.scope, budgetLimit, now),
           },
         }, { signal: activeWork.signal });
@@ -99,7 +99,7 @@ export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOpti
       const afterProviderMapping = options.serviceDb.getChannelMapping(options.scope.channelId);
       if (!isDiscordParticipantScopeAllowed(options.startup, options.scope, afterProviderMapping)) return "disabled";
       if (appraisal.kind === "unavailable") return "provider_unavailable";
-      if (appraisal.kind === "valid" && appraisal.proposal.confidence < (ambientSettings.ambientConfidenceFloor ?? 0.7)) {
+      if (appraisal.kind === "valid" && appraisal.proposal.confidence < (ambientSettings.ambientConfidenceFloor ?? DEFAULT_AMBIENT_CONFIDENCE_FLOOR)) {
         appraisal = { kind: "invalid", diagnostic: "provider confidence was below threshold" };
       }
       if (appraisal.kind === "valid" && !relationshipTargetsAuthorized(appraisal.proposal.relationshipProposals, context.transcript, roster)) {
@@ -113,11 +113,10 @@ export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOpti
         message: context.event,
         botUserId: options.botUserId,
         roster,
-        activeHumanIds,
         nowMs: now,
         observeOnly: work.observeOnly,
         ambientPityEnabled: ambientSettings.ambientPityEnabled ?? true,
-        confidenceFloor: ambientSettings.ambientConfidenceFloor ?? 0.7,
+        confidenceFloor: ambientSettings.ambientConfidenceFloor ?? DEFAULT_AMBIENT_CONFIDENCE_FLOOR,
         idleDecayTauMs: ambientSettings.ambientIdleDecayTauMs,
         pressureTauMs: ambientSettings.ambientPressureTauMs,
       });
@@ -155,6 +154,7 @@ export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOpti
         ...(budget ? { budget } : {}),
         ...(planned ? { plan: planFor(work.id, options.scope, work.eventId, bubbles!) } : {}),
         workId: work.id,
+        batchHighWatermark: { createdAtMs: work.createdAtMs, workId: work.id },
       });
       if (result === "idempotent") return "idle";
       return decision.audit.outcome === "invalid" ? "invalid" : planned ? "planned" : "observe";
@@ -237,7 +237,7 @@ function budgetRemaining(store: AdaptiveAmbientStore, scope: Scope, limit: numbe
 }
 function nextBudget(store: AdaptiveAmbientStore, scope: Scope, now: number) { const windowStartMs = hourStart(now); const current = store.budget(scope, BUDGET_KEY); return { key: BUDGET_KEY, count: current?.windowStartMs === windowStartMs ? current.count + 1 : 1, windowStartMs }; }
 function hourStart(now: number): number { return Math.floor(now / 3_600_000) * 3_600_000; }
-function planFor(workId: string, scope: Scope, eventId: string, chunks: readonly string[]) { return { id: `ambient:${scope.guildId}:${scope.channelId}:${eventId}`, workId, chunks: chunks.map((content, index) => ({ content, nonce: createHash("sha256").update(`ambient-plan-v1:${scope.guildId}:${scope.channelId}:${eventId}:${index}`).digest("hex").slice(0, 32) })) }; }
+function planFor(workId: string, scope: Scope, eventId: string, chunks: readonly string[]) { return { id: `ambient:${scope.guildId}:${scope.channelId}:${eventId}`, workId, chunks: chunks.map((content, index) => ({ content, nonce: createHash("sha256").update(`ambient-plan-v1:${scope.guildId}:${scope.channelId}:${eventId}:${index}`).digest("hex").slice(0, 24) })) }; }
 function personaFor(db: ServiceDatabase, profileId: string | null, globalPersona: string | undefined): string { return db.getProfile(profileId ?? "")?.soulSnippet?.trim() || globalPersona?.trim() || GENERIC_CONVERSATION_PERSONA; }
 function relationship(row: { user_id: string; rapport: number; familiarity: number; notes_json: string }): RelationshipContext { return { userId: row.user_id, rapport: row.rapport, familiarity: row.familiarity, notes: parseStringList(row.notes_json) }; }
 function parseRecord(value: string): Record<string, unknown> { try { const parsed: unknown = JSON.parse(value); return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as Record<string, unknown> : {}; } catch { return {}; } }