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

@@ -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");
@@ -103,7 +109,10 @@ describe("strict adaptive ambient appraisal provider", () => {
 
     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 } });
+    const soloAudience = { ...audience, activeHumanCount: 1 };
+    await ambientProvider(fetchImpl).appraise(request(soloAudience));
+    expect(JSON.parse(wireBody?.messages[1]?.content ?? "{}")).toMatchObject({ audience: soloAudience, participationPrior: { speak: 0.85, observe: 0.15 } });
   });
 
   it("turns every provider and contract failure into invalid audit input without leaking secrets", async () => {

service/src/adaptive-ambient-provider.ts

@@ -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.",
+      "When audience.activeHumanCount is at least 1, 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,16 @@ 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(request.audience.activeHumanCount) }),
     }),
   };
 }
 
+function participationPrior(activeHumanCount: number): { readonly speak: number; readonly observe: number } {
+  if (activeHumanCount >= 2) return { speak: 0.9, observe: 0.1 };
+  return activeHumanCount === 1 ? { speak: 0.85, observe: 0.15 } : { speak: 0.2, observe: 0.8 };
+}
+
 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: 0.7 * 0.75 + 1 * 0.25, 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();
 
@@ -173,7 +173,7 @@ describe("atomic adaptive ambient runtime", () => {
   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();
@@ -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 {
@@ -88,7 +89,7 @@ export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOpti
           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 +100,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)) {
@@ -117,7 +118,7 @@ export function createAdaptiveAmbientRuntime(options: AdaptiveAmbientRuntimeOpti
         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 +156,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 +239,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 {}; } }

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

@@ -59,6 +59,28 @@ describe("adaptive ambient persistence", () => {
     db.close();
   });
 
+  it("claims the latest actionable event as a fixed batch high-watermark", () => {
+    const fakeClock = clock(); const db = new service.ServiceDatabase(); const store = service.createAdaptiveAmbientStore(db, fakeClock.read);
+    let fence = store.acquireLease("discord-worker", "worker-a")!;
+    work(store, "work-old"); expect(store.claimWork("work-old", fence)).toBe(true);
+    fakeClock.advance(30_001); fence = store.acquireLease("discord-worker", "worker-a")!;
+    work(store, "work-latest");
+    expect(store.claimNextWork({ guildId: "g1", channelId: "c1" }, fence)).toBe("work-latest");
+    const watermark = store.work("work-latest")!;
+    fakeClock.advance(1); work(store, "work-next-tick");
+
+    expect(store.recordOutcome({
+      fence, eventId: "work-latest", scope: { guildId: "g1", channelId: "c1" }, outcome: "observe", workId: "work-latest",
+      batchHighWatermark: { createdAtMs: watermark.createdAtMs, workId: watermark.id }, state: { drive: 0.6, version: 1 },
+    })).toBe("applied");
+    expect(db.db.prepare("SELECT id,status FROM participant_event_work ORDER BY created_at_ms,id").all()).toEqual([
+      { id: "work-old", status: "observe" },
+      { id: "work-latest", status: "observe" },
+      { id: "work-next-tick", status: "pending" },
+    ]);
+    db.close();
+  });
+
   it("persists streak state and defaults stale null streaks to zero", () => {
     const fakeClock = clock(); const db = new service.ServiceDatabase(); const store = service.createAdaptiveAmbientStore(db, fakeClock.read);
     const fence = store.acquireLease("discord-worker", "worker-a")!;