Fix mock admin @all and in-place compaction in smoke (#917)

changelog.d/917.fixed.md

@@ -0,0 +1 @@
+A conversation that is still in the middle of its work stays on the same part after a checkpoint. The context meter was still counting messages the checkpoint had already replaced, so the part looked too long and moved on anyway ([#917](https://github.com/ThinkFlowLab/FlyBot/issues/917)).

core/dev/mock-llm.mjs

@@ -51,7 +51,7 @@
 import { createServer } from 'node:http'
 import { createHash } from 'node:crypto'
 import { appendFileSync, mkdirSync } from 'node:fs'
-import { dirname } from 'node:path'
+import { dirname, resolve } from 'node:path'
 import { fileURLToPath } from 'node:url'
 
 const PORT = Number(process.env.MOCK_PORT ?? 8765)
@@ -370,9 +370,13 @@ function decide(body) {
       .map(line => /^[^:\n]{1,40}: \D*?(\d+)/.exec(line)?.[1])
       .filter(Boolean).map(Number)
     const mentioner = /^\[Group chat [^\]]*· (\S+) mentioned you\]/.exec(userText)?.[1]
-    const peer = /^Members: the user, ([^,\n]+?)(?:,| and you)/m.exec(userText)?.[1]?.replace(/ \(admin\)$/, '')
+    // Who is here, and who leads, is the group's system-prompt section (#894).
+    // The turn's own text no longer repeats it. Older prompts still put the
+    // members line in the user message, so both are read.
+    const roster = `${system}\n${userText}`
+    const peer = /^Members: the user, ([^,\n]+?)(?:,| and you)/m.exec(roster)?.[1]?.replace(/ \(admin\)$/, '')
     const room = /^\[Group chat "([^"]+)"/.exec(userText)?.[1] ?? ''
-    const leads = /^Members: .* and you \(admin\)\./m.test(userText)
+    const leads = /^Members: .* and you \(admin\)\./m.test(roster)
     let ask
     if ((ask = /把\s*(\S+?)\s*拉进群/.exec(userLine))) return [tool('update_group', { group: room, add_members: [ask[1]] })]
     if ((ask = /把\s*(\S+?)\s*踢出群/.exec(userLine))) return [tool('update_group', { group: room, remove_members: [ask[1]] })]
@@ -727,4 +731,8 @@ const server = createServer((req, res) => {
   })
 })
 
-server.listen(PORT, '127.0.0.1', () => console.log(`ds-bot mock on http://127.0.0.1:${server.address().port}/v1`))
+// Tests import decide(); only a direct run should bind the port.
+const invokedDirectly = process.argv[1] !== undefined && fileURLToPath(import.meta.url) === resolve(process.argv[1])
+if (invokedDirectly) server.listen(PORT, '127.0.0.1', () => console.log(`ds-bot mock on http://127.0.0.1:${server.address().port}/v1`))
+
+export { decide }

core/src/host/parts.js

@@ -150,21 +150,44 @@ export function install(rt) {
   // A broken provider adapter must not crash every turn; estimate from the
   // derived messages instead, with one warning.
   let measureWarned = false
-  const tokensOf = (session) => {
+  const measuredOf = (session) => {
     try {
-      return ctx.get('tokenMeter')?.measure(session).totalTokens
+      return ctx.get('tokenMeter')?.measure(session)
     } catch (error) {
       if (!measureWarned) {
         measureWarned = true
         rt.warn('ds-bot: token measure failed; context sizes are estimated: %s', error)
       }
     }
+    return undefined
+  }
+  const tokensOf = (session) => {
+    const measured = measuredOf(session)
+    if (measured !== undefined) return measured.totalTokens
     try {
       return estimateTokens(session.deriveMessages().map(message => textOf(message.content)).join('\n'))
     } catch {
       return undefined
     }
   }
+  // The meter's total stays anchored on the provider usage from before a span
+  // was replaced, and the surface delta only subtracts a 4-characters-per-token
+  // estimate of the removed text. A denser report (the mock's 3 characters per
+  // token, or CJK on a real model) therefore still reads above the check line
+  // after the checkpoint has replaced the span. The heuristic size of what is
+  // left — the current surface plus the tool schemas, at that same density —
+  // is the prompt that remains. A reading with no such delta stays on the
+  // anchored total, which is what decides that the line was crossed.
+  const METER_CHARS_PER_TOKEN = 4
+  function pressureLeft(session) {
+    const measured = measuredOf(session)
+    if (measured === undefined) return tokensOf(session)
+    const anchored = measured.totalTokens
+    if (measured.baseline?.kind !== 'usage' || !(measured.surfaceDeltaTokens < 0) || typeof measured.surfaceTokens !== 'number') return anchored
+    const tools = session.requestHeader()?.tools
+    const toolTokens = Array.isArray(tools) && tools.length > 0 ? Math.ceil(JSON.stringify(tools).length / METER_CHARS_PER_TOKEN) : 0
+    return Math.min(anchored, measured.surfaceTokens + toolTokens)
+  }
   const clock = createClock(config.timeZone)
 
   // Lines for the model the Session last routed to; undefined while it has not
@@ -545,7 +568,7 @@ export function install(rt) {
     }
     // The prefix cache starts over anyway, so the memory section catches up.
     rt.renewMemory(sessionId)
-    const after = tokensOf(agent.session) ?? before
+    const after = pressureLeft(agent.session) ?? before
     partBase.set(agent.id, after)
     ctx.logger.info('ds-bot: %s condensed from %d to %d tokens (check line %d, hard line %d)', agent.id, before ?? -1, after ?? -1, lines.check, lines.hard)
     return after < lines.check

core/test/mock-group-roster.test.js

@@ -0,0 +1,38 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import { decide } from '../dev/mock-llm.mjs'
+
+// #894 put the members line in the system-prompt section. A group turn's own
+// text no longer says who leads, which is what "admin calls everyone" reads.
+const standing = [
+  'You are Chief, a Bot on the user\'s team.',
+  'You are the Main Bot.',
+  '',
+  '# This group chat',
+  'Members: the user, Writer, Coder and you (admin). Members speak one at a time; each message is posted before the next speaker starts.',
+].join('\n')
+
+const turn = [
+  '[Group chat "研发群" · your turn]',
+  '',
+  'This round so far, starting with the new message (oldest first):',
+  'User: 讨论',
+].join('\n')
+
+function groupTurn(system) {
+  return decide({
+    system,
+    tools: [{ name: 'update_group' }],
+    messages: [{ role: 'user', content: [{ type: 'text', text: turn }] }],
+  })
+}
+
+test('the mock treats the caller as admin from the system section, not the turn (#917)', () => {
+  const out = groupTurn(standing)
+  assert.equal(out[0].text, '@all 大家各说一句。')
+  // A member's section names someone else as admin, so the same turn is not an @all.
+  const member = standing.replace('and you (admin).', 'and you.').replace('Writer, Coder', 'Writer (admin), Coder')
+  assert.equal(groupTurn(member)[0].text, '我觉得可以。')
+  // The turn itself carries no members line; without the section the caller is not the admin.
+  assert.equal(groupTurn('You are Chief, a Bot on the user\'s team.')[0].text, '我觉得可以。')
+})

core/test/session-identity.test.js

@@ -49,13 +49,13 @@ function start({ config = {} } = {}) {
     summarize() {},
     async compactIfNeeded() {
       bench.compacted += 1
-      bench.tokens = 100
+      if (bench.holdTokens !== true) bench.tokens = 100
       return {}
     },
     async compactNow() {},
     async compactRegion() {},
   }
-  const ctx = fakeContext({ services: { compaction, tokenMeter: { measure: () => ({ totalTokens: bench.tokens }) } } })
+  const ctx = fakeContext({ services: { compaction, tokenMeter: { measure: () => ({ totalTokens: bench.tokens, ...bench.meter }) } } })
   async function* route(options) {
     bench.requests.push(options)
     if (options.sessionId === undefined) {
@@ -124,6 +124,48 @@ for (const sessionId of ['b2', 'g3']) {
   })
 }
 
+// #917: compaction replaces the span, but the meter's usage anchor only subtracts
+// its 4-characters-per-token estimate of what it removed. A denser provider
+// report (the mock's 3 characters per token) still reads above the check line.
+test('a checkpoint stays in place when the remaining surface fits and only the usage anchor reads high (#917)', async (t) => {
+  const { ctx, bench, stop } = start()
+  t.after(stop)
+  bench.tokens = 95_000
+  bench.holdTokens = true
+  bench.meter = { baseline: { kind: 'usage', tokens: 95_000 }, surfaceDeltaTokens: -40_000, surfaceTokens: 8_000 }
+  bench.respond = () => '{"action":"compact"}'
+  const session = {
+    id: 'b2', seq: 2,
+    requestHeader: () => ({ config: { provider: 'deepseek', model: 'v4' }, tools: [{ name: 'read' }] }),
+    deriveMessages: () => [], toolHistory: () => [],
+  }
+  const preStep = ctx.listeners.get('agent/pre-step').find(listener => /adoptCompaction/.test(String(listener)))
+  const next = { kind: 'continue' }
+  const result = await preStep({ agent: { id: 'b2', session }, messages: [], step: 2, signal: new AbortController().signal }, () => next)
+  assert.equal(result, next, 'the part switched even though the remaining surface fits under the check line')
+  assert.equal(bench.compacted, 1)
+})
+
+test('a checkpoint that leaves the surface over the check line still moves the part on (#917)', async (t) => {
+  const { ctx, bench, stop } = start()
+  t.after(stop)
+  bench.tokens = 95_000
+  bench.holdTokens = true
+  bench.meter = { baseline: { kind: 'usage', tokens: 95_000 }, surfaceDeltaTokens: -1_000, surfaceTokens: 90_000 }
+  bench.respond = () => '{"action":"compact"}'
+  const session = {
+    id: 'b2', seq: 2,
+    requestHeader: () => ({ config: { provider: 'deepseek', model: 'v4' } }),
+    deriveMessages: () => [], toolHistory: () => [],
+    // The step is already rejected; the switch job must not run far enough to touch a stub agent.
+    whenIdle: () => new Promise(() => {}),
+  }
+  const preStep = ctx.listeners.get('agent/pre-step').find(listener => /adoptCompaction/.test(String(listener)))
+  const next = { kind: 'continue' }
+  const result = await preStep({ agent: { id: 'b2', session }, messages: [], step: 2, signal: new AbortController().signal }, () => next)
+  assert.equal(result?.kind, 'reject', 'a surface that is still over the check line should move on')
+})
+
 test('an image in a Bot chat is read for that chat, and the chat gets the reading', async (t) => {
   const { ctx, bench, stop } = start()
   t.after(stop)

docs/subsystems/core-host.md

@@ -1,3 +1,3 @@
 ---
 subsystem: core-host
 owned_paths:
@@ -241,6 +241,7 @@
 - A friend's line in a Bot's own chat (#420): `deliver()` (`noteFriendLine`) stamps the receiving Bot with `friendLineAt` (published in the roster, so the save moves the revision) and keeps the senders seen per Bot in `state.remoteContacts[botId][handle]` (host only, capped at 200, dropped with the Bot). A sender's first line carries `source.remote.firstContact`. Group lines, and re-deliveries (`deferred`, `resumed`, `retried`, `resent`), stamp nothing. Covered by `core/test/flybot.test.js`.
 - **Pickup after a restart** (`resume.js`): every running Bot turn and every delivered-but-not-taken message is noted in `inflight.json`. At the next start a noted turn is picked up once only if a restart really cut it: the durable Session log's `turn/end` for that turn is `interrupted` (or an abort caused by `disposed`, or there is no end). A turn that finished while this plugin was not listening — `completed`, `blocked`, `max-tokens`, `forked`, or aborted by the user / hook / parent — is not repeated; a `turn/start` seen after the start means the noted turn ended too. Notes from a build before the record carried `turn` keep the old behaviour. A pickup is confirmed before its note is dropped: the notes read at start are seeded into the live `turns`/`queued` (so the file keeps everything still owed), a turn's note stays while its pickup runs and is dropped only once handed over (unless the resumed turn has written its own note first), and a queued message whose resend fails is kept for the next start, an exchange is noted once (a resend delivers the `exchangeId` the notes already hold, and noting it again left a copy behind after a turn took it up, so the start after that sent the message once more), and a turn taking up a message forgets every note of its exchange; a crash or a failed delivery during the pickup no longer loses the work — a pickup is at-least-once, not once: a crash between handing the turn over and the write that drops its note repeats that pickup once, because the `taken` set and the inbox that would tell the next start otherwise are process-local. That is the safe side of the trade, a repeated turn the user can see instead of work the user was told about going silently dead. Covered by `core/test/follow-through.test.js`.
 - A repeated Bot-to-Bot line becomes one turn (#641): `deliver()` keys the last 60 s of Bot-to-Bot text by sender + recipient Session + the text with its whitespace collapsed (`DEDUPE_MS`, `duplicateDelivery`), drops a repeat before it is recorded as an exchange or handed to the Session, and journals the drop as `delivery: dedupe <sender> -> <session>` in the host warning log. The key includes the Session, so the same line to another Session (a group copy, another Bot) is its own delivery; `deferred` / `resumed` / `retried` / `resent` carry on a message the Session already has and are never judged, and a friend's line, the user's own message and a group round are not Bot-to-Bot. Routing (parent and direct children) and the existing unread-queue refusal are unchanged, and the map is process-local and capped at 500 entries. Covered by `core/test/repeated-delivery.test.js`.
+- **A checkpoint is judged by the prompt that is left** (`parts.js` `pressureLeft`): the token meter stays anchored on the provider usage from before the span was replaced, and its surface delta only subtracts a 4-characters-per-token estimate of the removed text. A denser report — the mock's 3 characters per token, or CJK on a real model — can therefore still read above the check line after the checkpoint has replaced the span, and the part would move on even though the steward said to condense in place. After a compaction, the part stays when the heuristic size of the remaining surface plus the tool schemas is under the check line. A surface that is still over it moves on, as before. The decision to cross the line in the first place still uses the anchored total (#917).
 - **A part is committed before its handoff runs** (`parts.js` `moveOn`): the Bot's `sessionId` / `room.sessions`, the new segment and the `rescued` mark are saved first, so the turn the handoff starts can never see an uncommitted part — a failure handler that did would start a second switch against it and lose the previous part's note, and the rescue would not know it may only happen once. The handoff package is written next; if that fails, the pointer is rolled back (best effort saved, the fresh Session archived, the rescued mark dropped), so the old part is current and the finalizer hands the input on there. Two windows stay open, and both are written down rather than discovered later: a crash after that save and before the handoff package is flushed leaves the fresh part current with no package — the stopped turn's input still goes to it, so no work is lost, but the note is, and the rollback never runs; and the rollback's own save is best effort (`rt.save().catch(() => {})`), so a failing rollback leaves the file showing the fresh part while memory holds the old one, and the next start behaves like the first window. Neither is a reason to clear the pointer before the save: a note lost this way is a part without a handoff note, while clearing first loses the handoff package itself. The move is marked done only after the stopped turn's input and the old inbox messages are forwarded; a switch that never starts (no place, another switch, a read-only host) lets the step continue instead of rejecting its messages, which DSH has already taken out of the inbox. An armed scheduled task keeps the earlier Session active and DSH refuses to archive it, so the switch keeps the `moveSchedules` promise and waits for the move before archiving the earlier part; when the move failed and the refusal reports `schedule` as the only activity, the plain archive is retried with `{ stopActivity: true }` (the way a deletion does, `roster.js:450`), so a finished part is never left active with a task nobody sees; a refusal caused by a job, turn or subagent is not retried, because `stopActivity` would stop that too (#896). Covered by `core/test/work-dir.test.js`, `core/test/part-archive.test.js` and `core/test/bot-messages.test.js`.
 
 ### Connectors: GitHub and the Composio gateway (#311)

docs/subsystems/ds-bot-core.md

@@ -48,7 +48,7 @@ holds. Task bookkeeping does not imply runtime execution or tool permission.
 
 - `core/index.js` — plugin server.
 - `core/client.js` — built browser bundle (generated; never hand-edit).
-- `dev/bench.sh`, `dev/smoke.mjs` — test bench.
+- `dev/bench.sh`, `dev/smoke.mjs` — test bench. The mock reads who leads a group from the system prompt (`Members: … and you (admin).`), which is where the standing section puts that line (#894). A turn's own text no longer carries it, so "admin calls everyone" in the smoke is decided from the section (#917).
 
 ## Invariants