Fix Mesh synchronization after stale sequence rejection

docs/reporting-agent.md

@@ -207,3 +207,9 @@ Confirm that `MESH_HUB_URL` points to the public ingress and that its `/healthz`
 ### No new data appears
 
 Confirm that the process is still running, the local Codex sources are readable, the hub is reachable, and `/admin` shows a recent machine update. Keep the source directories read-only; do not broaden access to the entire `.codex` directory.
+
+### The hub rejects a sequence as already processed
+
+A restored state file, or another execution context using the same machine identity, can leave the local counter behind the hub. The agent retries a recognized replay rejection once with a counter based on the current time, persisting it before sending. The hub still verifies the signature and rejects any counter that is not newer. Other conflicts and authentication errors are not retried this way.
+
+Keep only one sender for each identity. Stop the supervised agent before running diagnostic signed requests with its state, and use the same execution context and state file. If the retry also fails, check the machine clock and look for another sender; do not reset the hub counter or re-enroll the machine to bypass replay protection.

src/mesh-agent.mjs

@@ -147,7 +147,7 @@ export class MeshAgent {
     return result;
   }
 
-  async sendSigned(pathname, payload) {
+  async sendSigned(pathname, payload, recoverReplay = true) {
     const sequence = this.state.sequence + 1;
     const envelope = createSignedEnvelope({
       nodeId: this.state.nodeId,
@@ -165,8 +165,20 @@ export class MeshAgent {
       headers: this.requestHeaders(),
       body: canonicalJson(envelope),
     });
-    const result = await responseJson(response);
-    return result;
+    try {
+      return await responseJson(response);
+    } catch (error) {
+      const replay = error.status === 409 && (error.code === "mesh_replay"
+        || (!error.code && ["Séquence déjà traitée.", "Séquence Mesh déjà traitée."].includes(error.message)));
+      // A restored or isolated state file can lag a counter advanced by another
+      // instance of this identity. Skip to the clock once, never reset the hub's
+      // replay protection or retry unrelated conflicts. sendSigned persists the
+      // new reservation before transmitting, including if this retry fails.
+      const clockSequence = Date.now();
+      if (!recoverReplay || !replay || clockSequence <= sequence) throw error;
+      this.state.sequence = clockSequence - 1;
+      return this.sendSigned(pathname, payload, false);
+    }
   }
 
   async runSync(data) {

test/mesh-agent.test.mjs

@@ -205,6 +205,55 @@ test("agent never reuses a reserved sequence after an interrupted request", asyn
   assert.deepEqual(sequences, [1, 2]);
 });
 
+test("agent recovers a stale counter once without weakening hub replay protection", async (t) => {
+  for (const body of [{ error: "Séquence déjà traitée." }, { error: "Séquence Mesh déjà traitée.", code: "mesh_replay" }]) {
+    await t.test(body.code || "Sites legacy response", async () => {
+      const directory = await mkdtemp(path.join(os.tmpdir(), "codex-mesh-replay-"));
+      const requests = [];
+      const floor = Date.now() - 1000;
+      const agent = new MeshAgent({
+        hubUrl: "https://mesh.example", statePath: path.join(directory, "agent.json"),
+        fetchImpl: async (_url, options) => {
+          const envelope = JSON.parse(options.body);
+          requests.push(envelope);
+          const persisted = JSON.parse(await readFile(agent.statePath, "utf8"));
+          assert.equal(persisted.sequence, envelope.sequence, "reserve before every transmission");
+          return envelope.sequence <= floor ? Response.json(body, { status: 409 }) : Response.json({ accepted: true });
+        },
+      });
+      await agent.load(); agent.state.nodeId = "node_test";
+      const payload = { kind: "read", requestVersion: 1 };
+      await agent.sendSigned("/api/mesh/usage", payload);
+      assert.equal(requests.length, 2);
+      assert.equal(requests[0].sequence, 1);
+      assert.ok(requests[1].sequence > floor);
+      assert.deepEqual(requests[1].payload, payload);
+      const { verifySignedEnvelope } = await import("../src/mesh-protocol.mjs");
+      verifySignedEnvelope(requests[1], agent.state.publicKey);
+      const recovered = agent.state.sequence;
+      await agent.sendSigned("/api/mesh/usage", payload);
+      assert.equal(requests[2].sequence, recovered + 1);
+    });
+  }
+});
+
+test("replay recovery is bounded and does not rewind a counter ahead of the clock", async () => {
+  for (const initial of [0, Date.now() + 60_000]) {
+    const directory = await mkdtemp(path.join(os.tmpdir(), "codex-mesh-replay-limit-"));
+    let attempts = 0;
+    const agent = new MeshAgent({
+      hubUrl: "https://mesh.example", statePath: path.join(directory, "agent.json"),
+      fetchImpl: async () => { attempts++; return Response.json({ error: "Séquence déjà traitée." }, { status: 409 }); },
+    });
+    await agent.load(); agent.state.nodeId = "node_test"; agent.state.sequence = initial;
+    await assert.rejects(agent.sendSigned("/api/mesh/usage", { kind: "read", requestVersion: 1 }), /Séquence déjà traitée/);
+    assert.equal(attempts, initial ? 1 : 2);
+    assert.ok(agent.state.sequence > initial);
+    const persisted = JSON.parse(await readFile(agent.statePath, "utf8"));
+    assert.equal(persisted.sequence, agent.state.sequence);
+  }
+});
+
 test("hub rejects unexpected private fields before storing a snapshot", async () => {
   const store = new MeshHubStore();
   const enrollment = await store.createEnrollment();