fix(connector): recover corrupt admission journals safely

desktop-connector/src/admissions.mjs

@@ -0,0 +1,69 @@
+import { readFile, writeFile, rename, chmod } from "node:fs/promises"
+import { randomUUID } from "node:crypto"
+
+const record = value => value !== null && typeof value === "object" && !Array.isArray(value)
+const invalid = message => Object.assign(new Error(message), { code: "INVALID_ADMISSIONS", mobileStorage: true })
+
+async function recoveryBefore(path) {
+  try {
+    const value = JSON.parse(await readFile(path + ".recovery", "utf8"))
+    if (!record(value) || !Number.isSafeInteger(value.before) || value.before < 0)
+      throw invalid("Invalid mobile admission recovery boundary")
+    return value.before
+  } catch (error) {
+    if (error.code === "ENOENT") return undefined
+    if (error instanceof SyntaxError) throw invalid("Invalid mobile admission recovery boundary")
+    error.mobileStorage = true
+    throw error
+  }
+}
+
+async function journal(path) {
+  try {
+    const value = JSON.parse(await readFile(path, "utf8"))
+    if (!record(value) || !record(value.admissions) || !record(value.selections) ||
+      Object.entries(value.admissions).some(([key, item]) =>
+        !/^[A-Za-z0-9_-]+\.msg_[A-Za-z0-9_-]+$/.test(key) || !record(item) ||
+        !/^[a-f0-9]{64}$/.test(item.hash) || !["forwarding", "admitted"].includes(item.state)) ||
+      Object.values(value.selections).some(item => !record(item)) ||
+      (value.recoverySessions !== undefined && (!Array.isArray(value.recoverySessions) ||
+        value.recoverySessions.some(id => typeof id !== "string" || !/^ses_[a-f0-9]{32}$/.test(id)))))
+      throw invalid("Invalid mobile admission journal")
+    return value
+  } catch (error) {
+    if (error.code === "ENOENT") return { admissions: {}, selections: {} }
+    if (error instanceof SyntaxError) throw invalid("Invalid mobile admission journal")
+    error.mobileStorage = true
+    throw error
+  }
+}
+
+export async function openAdmissions(path) {
+  const before = await recoveryBefore(path)
+  return { state: await journal(path), recovered: before !== undefined }
+}
+
+// Reset never discards a healthy journal: it may contain an uncertain delivery.
+// Persist the fence BEFORE quarantine so a crash cannot reopen an empty journal
+// without its duplicate-delivery protection. Healthy resets leave it unchanged;
+// another corrupt journal must fence deliveries made since the previous recovery.
+export async function resetAdmissions(path) {
+  const before = await recoveryBefore(path)
+  try {
+    await journal(path)
+    return
+  } catch (error) {
+    if (error.code !== "INVALID_ADMISSIONS") throw error
+  }
+  try {
+    const temporary = path + ".recovery." + randomUUID()
+    await writeFile(temporary, JSON.stringify({ before: Math.max(before ?? 0, Date.now()) }), { flag: "wx", mode: 0o600 })
+    await rename(temporary, path + ".recovery")
+    // A damaged file can contain previous selections; keep its backup private.
+    await chmod(path, 0o600)
+    await rename(path, path + ".corrupt-" + randomUUID())
+  } catch (error) {
+    error.mobileStorage = true
+    throw error
+  }
+}

desktop-connector/src/desktop-access.mjs

@@ -4,6 +4,7 @@ import QRCode from "qrcode"
 import { createGateway } from "./gateway.mjs"
 import { connectTunnel, removeTunnel } from "./tunnel.mjs"
 import { openDevices, resetDevices } from "./devices.mjs"
+import { resetAdmissions } from "./admissions.mjs"
 
 export async function createMobileAccess(options) {
   const root = join(options.userDataPath, "mobile-access")
@@ -136,7 +137,7 @@ export async function createMobileAccess(options) {
     } catch (error) {
       if (token === epoch) {
         status = "error"
-        errorKey ??= "mobile.error.connection"
+        errorKey ??= error.mobileStorage ? "mobile.error.storage" : "mobile.error.connection"
       }
       throw error
     } finally {
@@ -190,7 +191,15 @@ export async function createMobileAccess(options) {
       await shutdown(true)
     } else if (input.type === "reset") {
       await shutdown(true)
-      devices = (await resetDevices(storePath)).list()
+      try {
+        await resetAdmissions(join(root, "admissions.json"))
+        devices = (await resetDevices(storePath)).list()
+      } catch (error) {
+        status = "error"
+        errorKey = "mobile.error.storage"
+        emit()
+        throw error
+      }
       await (options.removeTunnel ?? removeTunnel)({ storePath: join(root, "tunnel-bharatcode-ai") })
     } else if (input.type === "approve") {
       if (!gateway) throw new Error("Remote access disabled")

desktop-connector/src/desktop-api.mjs

@@ -1,6 +1,7 @@
-import { readFile, writeFile, rename } from 'node:fs/promises'
-import { createHash } from 'node:crypto'
+import { writeFile, rename } from 'node:fs/promises'
+import { createHash, randomUUID } from 'node:crypto'
 import { setTimeout as delay } from 'node:timers/promises'
+import { openAdmissions } from './admissions.mjs'
 
 const fail = (status, message) => Object.assign(new Error(message), { status })
 const normalizedPermission = item => ({ ...item, action: item.permission, resources: item.patterns ?? [] })
@@ -16,8 +17,7 @@ export function desktopMessage(item) {
 // Desktop's default runtime serves native session metadata but continues chats
 // through its compatibility engine. Keep that same engine and message tables.
 export async function createDesktopAPI({ request, storePath }) {
-  let state = { admissions: {}, selections: {} }
-  try { state = { ...state, ...JSON.parse(await readFile(storePath,'utf8')) } } catch (error) { if(error.code !== 'ENOENT') throw error }
+  const { state, recovered } = await openAdmissions(storePath)
   let writes = Promise.resolve()
   const save = () => {
     const json = JSON.stringify(state)
@@ -43,6 +43,20 @@ export async function createDesktopAPI({ request, storePath }) {
   }
   async function handle(method, raw, body, signal) {
     const url = new URL(raw,'http://localhost'), path = url.pathname
+    if (method === 'POST' && path === '/api/session' && recovered) {
+      // Session creation can adopt a caller-supplied identity. During recovery
+      // only our own fresh random identity proves that a session has no lost
+      // admissions. Wall-clock timestamps cannot prove this after clock changes.
+      if (!body || typeof body !== 'object' || Array.isArray(body) || body.id !== undefined)
+        throw fail(400,'A fresh conversation identity is required after mobile history recovery')
+      if ((state.recoverySessions?.length ?? 0) >= 10000) throw fail(507,'Mobile recovery conversation history is full')
+      const id = 'ses_' + randomUUID().replaceAll('-','')
+      const result = await request(raw,'POST',{...body,id},signal)
+      if (result.data?.data?.id !== id) throw fail(502,'Desktop did not create the requested conversation')
+      state.recoverySessions = [...(state.recoverySessions ?? []),id]
+      await save()
+      return result
+    }
     const match = /^\/api\/session\/([A-Za-z0-9_-]+)\/(message|prompt|interrupt|permission|question|model|agent)$/.exec(path)
     if (method==='GET' && path==='/api/session') {
       const result = await request(raw,'GET',undefined,signal)
@@ -116,6 +130,16 @@ export async function createDesktopAPI({ request, storePath }) {
         }
         throw fail(409,'Desktop delivery is unconfirmed. Check the conversation on your computer before continuing.')
       }
+      if (recovered && !state.recoverySessions?.includes(id)) {
+        // With the old journal lost, absence in Desktop is not proof that an
+        // old request was never forwarded. Only our durably recorded fresh
+        // conversation identities can submit without that lost history.
+        if (await stored(id,body,info.directory,signal)) {
+          state.admissions[key]={hash,state:'admitted'}; await save()
+          return {data:{data:{id:body.id}}}
+        }
+        throw fail(409,'Mobile delivery history was recovered. Check this conversation on your computer; start a new conversation on your phone to send safely.')
+      }
       if(Object.keys(state.admissions).length>=10000) throw fail(507,'Mobile admission history is full')
       const operation=(async()=>{
         const selection=state.selections[id], model=selection?.model ?? info.model

desktop-connector/test/desktop-access.test.mjs

@@ -1,12 +1,90 @@
 import { test } from "node:test"
 import assert from "node:assert/strict"
-import { mkdtemp, rm, mkdir, writeFile, readFile, readdir } from "node:fs/promises"
+import { mkdtemp, rm, mkdir, writeFile, readFile, readdir, stat } from "node:fs/promises"
 import { join } from "node:path"
 import { tmpdir } from "node:os"
 import { createMobileAccess } from "../src/desktop-access.mjs"
 import QRCode from "qrcode"
 import { PNG } from "pngjs"
 
+for (const damaged of ['{"admissions":', '{"admissions":null,"selections":{}}', '{"admissions":{},"selections":[]}']) {
+  test(`explicit reset quarantines invalid admissions and allows pairing: ${damaged}`, async (t) => {
+    const directory = await mkdtemp(join(tmpdir(), "bc-admissions-reset-"))
+    t.after(() => rm(directory, { recursive: true, force: true }))
+    const root = join(directory, "mobile-access")
+    await mkdir(root)
+    await writeFile(join(root, "admissions.json"), damaged)
+    const options = {
+      userDataPath: directory,
+      getConnection: () => ({ url: "http://127.0.0.1:4096", password: "private" }),
+      removeTunnel: async () => {},
+      tunnelFactory: async () => ({ origin: "https://pc.tunnel.bharatcode.ai", events: (async function* () {})(), close: async () => {} }),
+    }
+    const access = await createMobileAccess(options)
+    t.after(() => access.close())
+    await assert.rejects(access.action({ type: "pair" }))
+    assert.equal(access.snapshot().errorKey, "mobile.error.storage")
+    assert.equal(await readFile(join(root, "admissions.json"), "utf8"), damaged)
+    await access.action({ type: "reset" })
+    const backups = (await readdir(root)).filter(name => name.startsWith("admissions.json.corrupt-"))
+    assert.equal(backups.length, 1)
+    assert.equal(await readFile(join(root, backups[0]), "utf8"), damaged)
+    if (process.platform !== "win32") assert.equal((await stat(join(root, backups[0]))).mode & 0o777, 0o600)
+    const fence = JSON.parse(await readFile(join(root, "admissions.json.recovery"), "utf8"))
+    assert.ok(Number.isSafeInteger(fence.before))
+    await access.action({ type: "pair" })
+    assert.equal(access.snapshot().status, "connected")
+    await access.action({ type: "reset" })
+    assert.deepEqual(JSON.parse(await readFile(join(root, "admissions.json.recovery"), "utf8")), fence)
+    await access.close()
+    const reopened = await createMobileAccess(options)
+    t.after(() => reopened.close())
+    await reopened.action({ type: "pair" })
+    assert.equal(reopened.snapshot().status, "connected")
+  })
+}
+
+test("reset preserves a healthy forwarding journal byte-for-byte", async (t) => {
+  const directory = await mkdtemp(join(tmpdir(), "bc-admissions-healthy-"))
+  t.after(() => rm(directory, { recursive: true, force: true }))
+  const root = join(directory, "mobile-access")
+  await mkdir(root)
+  const original = JSON.stringify({ admissions: { "ses_one.msg_one": { hash: "a".repeat(64), state: "forwarding" } }, selections: {} })
+  await writeFile(join(root, "admissions.json"), original)
+  const access = await createMobileAccess({ userDataPath: directory, getConnection: () => undefined, removeTunnel: async () => {} })
+  t.after(() => access.close())
+  await access.action({ type: "reset" })
+  assert.equal(await readFile(join(root, "admissions.json"), "utf8"), original)
+  assert.equal((await readdir(root)).some(name => name.includes(".corrupt-")), false)
+})
+
+test("reset reports storage failure instead of success when admissions cannot be read", async (t) => {
+  const directory = await mkdtemp(join(tmpdir(), "bc-admissions-unreadable-"))
+  t.after(() => rm(directory, { recursive: true, force: true }))
+  const root = join(directory, "mobile-access")
+  await mkdir(join(root, "admissions.json"), { recursive: true })
+  const access = await createMobileAccess({ userDataPath: directory, getConnection: () => undefined, removeTunnel: async () => {} })
+  t.after(() => access.close())
+  await assert.rejects(access.action({ type: "reset" }))
+  assert.equal(access.snapshot().status, "error")
+  assert.equal(access.snapshot().errorKey, "mobile.error.storage")
+  assert.equal((await readdir(root)).some(name => name.includes(".corrupt-")), false)
+})
+
+test("a second damaged journal advances the recovery fence to cover newer uncertain deliveries", async (t) => {
+  const directory = await mkdtemp(join(tmpdir(), "bc-admissions-second-reset-"))
+  t.after(() => rm(directory, { recursive: true, force: true }))
+  const root = join(directory, "mobile-access")
+  await mkdir(root)
+  await writeFile(join(root, "admissions.json.recovery"), '{"before":1000}')
+  await writeFile(join(root, "admissions.json"), '{"admissions":')
+  const access = await createMobileAccess({ userDataPath: directory, getConnection: () => undefined, removeTunnel: async () => {} })
+  t.after(() => access.close())
+  const started = Date.now()
+  await access.action({ type: "reset" })
+  assert.ok(JSON.parse(await readFile(join(root, "admissions.json.recovery"), "utf8")).before >= started)
+})
+
 test("pairing QR preserves a four-module clear border for physical scanners", async (t) => {
   const directory = await mkdtemp(join(tmpdir(), "bc-access-qr-margin-"))
   t.after(() => rm(directory, { recursive: true, force: true }))

desktop-connector/test/desktop-api.test.mjs

@@ -1,10 +1,64 @@
 import test from 'node:test'
 import assert from 'node:assert/strict'
-import { mkdtemp, rm } from 'node:fs/promises'
+import { mkdtemp, rm, writeFile } from 'node:fs/promises'
 import { join } from 'node:path'
 import { tmpdir } from 'node:os'
 import { createDesktopAPI } from '../src/desktop-api.mjs'
 
+for (const created of [undefined, 999, 1000, Date.now() + 3_600_000]) {
+  test(`recovery fence blocks unknown old delivery across restart, created=${created}`, async t => {
+    const root=await mkdtemp(join(tmpdir(),'bc-recovery-fence-'));t.after(()=>rm(root,{recursive:true,force:true}))
+    const storePath=join(root,'admissions.json')
+    // Simulate a crash after quarantine but before the replacement journal was written.
+    await writeFile(storePath+'.recovery',JSON.stringify({before:1000}))
+    let submissions=0
+    const request=async path=>{
+      if(path==='/session/ses_one')return {data:{directory:'/work',time:{created}}}
+      if(path.includes('/prompt_async')){submissions++;return {data:undefined}}
+      throw Object.assign(new Error('not stored'),{status:404})
+    }
+    for(let i=0;i<2;i++){
+      const api=await createDesktopAPI({request,storePath})
+      await assert.rejects(api.handle('POST','/api/session/ses_one/prompt',{id:'msg_one',prompt:{text:'Do not duplicate'}}),/recovered.*new conversation/i)
+    }
+    assert.equal(submissions,0)
+  })
+}
+
+test('recovered history acknowledges stored messages and permits only connector-created conversations',async t=>{
+  const root=await mkdtemp(join(tmpdir(),'bc-recovery-proof-'));t.after(()=>rm(root,{recursive:true,force:true}))
+  const storePath=join(root,'admissions.json')
+  await writeFile(storePath+'.recovery',JSON.stringify({before:1000}))
+  let submissions=0, fresh
+  const request=async(path,method,body)=>{
+    if(path==='/api/session' && method==='POST'){fresh=body.id;return {data:{data:{id:fresh}}}}
+    if(path==='/session/ses_old')return {data:{directory:'/work',time:{created:900}}}
+    if(path==='/session/'+fresh)return {data:{directory:'/work',time:{created:0}}}
+    if(path.includes('/prompt_async')){submissions++;return {data:undefined}}
+    return {data:{info:{id:'msg_one',role:'user'},parts:[{type:'text',text:'Only once'}]}}
+  }
+  const api=await createDesktopAPI({request,storePath})
+  const prompt={id:'msg_one',prompt:{text:'Only once'}}
+  await api.handle('POST','/api/session/ses_old/prompt',prompt)
+  assert.equal(submissions,0)
+  await assert.rejects(api.handle('POST','/api/session/ses_old/prompt',{...prompt,prompt:{text:'Different'}}),/identity/)
+  await assert.rejects(api.handle('POST','/api/session',{id:'ses_old'}),/identity/i)
+  const created=await api.handle('POST','/api/session',{location:{directory:'/work'}})
+  assert.match(created.data.data.id,/^ses_[a-f0-9]{32}$/)
+  await api.handle('POST',`/api/session/${fresh}/prompt`,prompt)
+  assert.equal(submissions,1)
+  const reopened=await createDesktopAPI({request,storePath})
+  await reopened.handle('POST',`/api/session/${fresh}/prompt`,prompt)
+  assert.equal(submissions,1)
+})
+
+test('malformed recovery fence fails closed instead of forgetting prior quarantine',async t=>{
+  const root=await mkdtemp(join(tmpdir(),'bc-recovery-invalid-'));t.after(()=>rm(root,{recursive:true,force:true}))
+  const storePath=join(root,'admissions.json')
+  await writeFile(storePath+'.recovery','{"before":null}')
+  await assert.rejects(createDesktopAPI({request:()=>{throw new Error('must not contact sidecar')},storePath}),/recovery/i)
+})
+
 test('stored admission is acknowledged while the upstream async HTTP acknowledgement is still pending', async t => {
   const root = await mkdtemp(join(tmpdir(),'bc-admission-early-'))
   t.after(()=>rm(root,{recursive:true,force:true}))