Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions changelog.d/chat-waiting-copy-send-stamp.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
### Fixed
- A message sent from Chat no longer stays faded beside its own turn after the PC's clock was set
back, and another device's identical message sent just before it no longer clears its waiting copy
when the chat had not read the pane for a long time.
22 changes: 16 additions & 6 deletions server/core/pending.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ const isClient = (owner: PendingOwner): owner is Client => owner !== FIRST_MESSA
/** a first message's lease is the delivering connection's, taken at each attempt: none is kept on the record */
type PendingItem = PendingRecord<PendingOwner, PendingLease | null>;
/** `refusal`: a queued request's own input_draft refusal, kept for its submit-result's draft actions; never sent as is */
type SubmitReply = { ok: boolean; pending?: PendingMessage; code?: string; message?: string; refusal?: { draft: string | null; box: string | null; identity: PendingIdentity } };
type SubmitReply = { ok: boolean; pending?: PendingMessage; sending_at?: string; code?: string; message?: string; refusal?: { draft: string | null; box: string | null; identity: PendingIdentity } };
type PiTypedReceipt = { owner: Client; resumeKey?: string; message: PendingMessage; match: string; template: boolean; identity: PendingIdentity; anchor: PiDeliveryAnchor | null };
/** a first message refused before anything was typed because its connection could not type it yet stays queued for its next attach or turn */
const FIRST_MESSAGE_RETRY_CODES = new Set(["input_not_ready", "pending_lease_lost", "disconnected", "read_only"]);
Expand Down Expand Up @@ -1012,8 +1012,9 @@ export function createPendingDelivery({ options, clients, attachments, sight, se
const pendingFrames = {
async submit(client: Client, message: Frame<"submit">): Promise<void> {
// every submit is answered: the composer keeps its text until it hears back
const result = (ok: boolean, code?: string, text?: string, accepted?: PendingMessage, draft?: string | null, box?: string | null, typedNow = false) => send(client, {
const result = (ok: boolean, code?: string, text?: string, accepted?: PendingMessage, draft?: string | null, box?: string | null, typedNow = false, sendingAt?: string) => send(client, {
type: "submit-result", id: message.id, pane_id: message.pane_id, ok, ...(accepted ? { pending: accepted } : {}), ...(typedNow ? { typed_now: true as const } : {}), ...(code ? { code, message: text } : {}),
...(ok && sendingAt ? { sending_at: sendingAt } : {}),
...(draft ? { draft } : {}), ...(!draft && typeof box === "string" ? { box } : {}),
});
if (!Number.isSafeInteger(message.id) || typeof message.pane_id !== "string" || !message.pane_id
Expand Down Expand Up @@ -1095,7 +1096,12 @@ export function createPendingDelivery({ options, clients, attachments, sight, se
// already waiting its turn here, this one takes its place behind it instead.
// A held message is never typed by arriving after the turn: it waits for Send now.
// Queued, it never clears a stopped send Claude put back in its box, as the drain does not (#2546).
if (message.delivery === "queue" && !pending.waiting(message.pane_id)) return dispatchPendingText(client, message.pane_id, text, lease!, context.identity, false, arrivedAt, undefined, false, message.id, agentOnly, true);
if (message.delivery === "queue" && !pending.waiting(message.pane_id)) {
// before anything is typed: the message's turn is written after this on this PC's clock (src/lib/sentWaiting.ts)
const sendingAt = new Date().toISOString();
const typed = await dispatchPendingText(client, message.pane_id, text, lease!, context.identity, false, arrivedAt, undefined, false, message.id, agentOnly, true);
return typed.ok ? { ...typed, sending_at: sendingAt } : typed;
}
}
pending.observe(message.pane_id, held && !context.working ? "working" : context.pane.agent_status, mark);
// hold: kept for the owner's Send now, never sent by the turn's end
Expand All @@ -1113,11 +1119,14 @@ export function createPendingDelivery({ options, clients, attachments, sight, se
confirmations.set(message.pane_id, refusal);
}
// typed by this very request: a replayed id, or a row gone since (sent or discarded), answers without it
const typedNow = !previous && answer.ok && answer.pending === undefined;
result(answer.ok, answer.code, answer.message, accepted, refusal?.draft ?? null, refusal && refusal.box !== null ? shownInputBox(refusal.box) : null,
!previous && answer.ok && answer.pending === undefined);
typedNow, typedNow ? answer.sending_at : undefined);
drainPending(message.pane_id);
return;
}
// the PC's clock before this request types anything, for the browser's waiting copy (src/lib/sentWaiting.ts)
let sendingAt: string | undefined;
const dispatch = async () => {
// held while this waited its turn (the attach was refused after the check above)
if (attachments.get(message.pane_id)?.held) throw new HerdrError("attach_held", ATTACH_HELD_MESSAGE);
Expand All @@ -1131,6 +1140,7 @@ export function createPendingDelivery({ options, clients, attachments, sight, se
try {
const receiptText = identity === null ? undefined : piReceiptText(identity, framed.text);
const piAnchor = identity === null ? null : await preparePiReceipt(message.pane_id, identity).catch(() => null);
sendingAt = new Date().toISOString();
await submitText(message.pane_id, framed.text, framed.payload, arrivedAt, message.typed === true, () => authorizeSocket(client), message.clear_draft,
draftConfirmations.get(client)?.get(message.pane_id), message.send_anyway === true, agentOnly);
if (identity !== null && receiptText !== undefined) {
Expand All @@ -1147,12 +1157,12 @@ export function createPendingDelivery({ options, clients, attachments, sight, se
try { await dispatch(); typedNow = true; return { ok: true }; }
catch (error) { return { ok: false, ...pendingFault(error) }; }
}, message.resend_generation));
result(answer.ok, answer.code, answer.message, undefined, undefined, undefined, answer.ok && typedNow);
result(answer.ok, answer.code, answer.message, undefined, undefined, undefined, answer.ok && typedNow, typedNow ? sendingAt : undefined);
return;
}
await serialize(message.pane_id, dispatch);
if (message.clear_draft !== undefined || message.send_anyway) draftConfirmations.get(client)?.delete(message.pane_id);
result(true);
result(true, undefined, undefined, undefined, undefined, undefined, false, sendingAt);
} catch (error) {
if (message.clear_draft !== undefined || message.send_anyway) draftConfirmations.get(client)?.delete(message.pane_id);
// a refusal naming a draft confirms it for Clear it and send; one naming none, for Send anyway
Expand Down
18 changes: 18 additions & 0 deletions server/pending-input.contract.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1465,17 +1465,22 @@ describe("connection-owned pending input", () => {
it("marks a queued send typed at once as typed now, and never its replay or a discarded row's", async () => {
const f = await setup("typed-now");
await f.state("idle");
const asked = Date.now();
f.socket.send({ type: "submit", id: 1, pane_id: f.pane, text: "typed at once", payload: "unused\r", delivery: "queue" });
const fresh = await f.socket.result(1);
expect(fresh).toMatchObject({ ok: true, typed_now: true });
expect(fresh.pending).toBeUndefined();
// the PC's clock before it typed: the message's turn is written after it (src/lib/sentWaiting.ts)
expect(Date.parse(fresh.sending_at)).toBeGreaterThanOrEqual(asked);
expect(Date.parse(fresh.sending_at)).toBeLessThanOrEqual(Date.now());
await f.waitBytes(`${paste("typed at once")}\r`);
// the same id again: its answer is replayed and nothing is typed twice
const bytes = f.bytes(), from = f.socket.seen.length;
f.socket.send({ type: "submit", id: 1, pane_id: f.pane, text: "typed at once", payload: "unused\r", delivery: "queue" });
const replay = await f.socket.wait((frame) => frame.type === "submit-result" && frame.id === 1, from);
expect(replay.ok).toBe(true);
expect(replay.typed_now).toBeUndefined();
expect(replay.sending_at).toBeUndefined();
expect(f.bytes()).toBe(bytes);
// queued while the agent works, then discarded: its replay answers ok with no row, and was never typed
await f.state("working");
Expand All @@ -1489,9 +1494,22 @@ describe("connection-owned pending input", () => {
expect(discarded.ok).toBe(true);
expect(discarded.pending).toBeUndefined();
expect(discarded.typed_now).toBeUndefined();
expect(discarded.sending_at).toBeUndefined();
expect(f.bytes()).toBe(bytes);
}, 30_000);

it("stamps an immediate send's result with the PC's clock before it typed", async () => {
const f = await setup("immediate-stamp");
await f.state("idle");
const asked = Date.now();
f.socket.send({ type: "submit", id: 1, pane_id: f.pane, text: "sent at once", payload: paste("sent at once") });
const answer = await f.socket.result(1);
expect(answer.ok).toBe(true);
expect(Date.parse(answer.sending_at)).toBeGreaterThanOrEqual(asked);
expect(Date.parse(answer.sending_at)).toBeLessThanOrEqual(Date.now());
await f.waitBytes(`${paste("sent at once")}\r`);
}, 30_000);

it("takes no colors from a scrolled viewport: an older grey box with the typed words holds the message", async () => {
const f = await setup("draft-scrolled");
const rule = "\u2500".repeat(60);
Expand Down
7 changes: 5 additions & 2 deletions shared/protocol.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1282,8 +1282,11 @@ export type ServerMessage =
* `typed_now`: on a queued send, this request typed the message at once (the turn had ended
* before it arrived); on a held copy's resend (`resend_id`), this request typed it. A replayed id, a
* row sent or discarded since, or a copy another connection sent answers ok without it; an older
* server never sends it. */
| { type: "submit-result"; id: number; pane_id: string; ok: boolean; pending?: PendingMessage; typed_now?: true; code?: string; message?: string; draft?: string; box?: string }
* server never sends it.
* `sending_at`: on an ok for a message this request typed (an immediate send, or one with
* `typed_now`), the PC's clock (ISO 8601) before it typed anything, as a queued row's `sending_at`:
* the message's turn is written after it on that clock. An older server never sends it. */
| { type: "submit-result"; id: number; pane_id: string; ok: boolean; pending?: PendingMessage; typed_now?: true; sending_at?: string; code?: string; message?: string; draft?: string; box?: string }
| { type: "pending-result"; id: number; pane_id: string; pending_id: string; ok: boolean; code?: string; message?: string }
/** ok once herdr accepted the inserted bytes (`pane.send_text` answered), never that the program
* took or ran them; a refusal names why and nothing was typed, except `insert_failed`, whose bytes
Expand Down
7 changes: 3 additions & 4 deletions src/lib/machineSession.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import { afterSend, afterSettled, greetingMemory, rememberGreeting } from "./gre
import { paneLifetimes } from "./paneLifetime.ts";
import { setPasteFolder } from "./pasteFolders.ts";
import { pendingMessages, type PendingMessageStore } from "./pendingMessages.ts";
import { rememberWaitingSend, waitingBaseline, waitingKey, type WaitingBaseline } from "./sentWaiting.ts";
import { rememberWaitingSend, stampedBaseline, waitingBaseline, waitingKey, type WaitingBaseline } from "./sentWaiting.ts";
import { watchPageInUse } from "./presence.ts";
import { currentSurface } from "./telemetrySurface.ts";
import { HerdrSocket } from "./ws.ts";
Expand Down Expand Up @@ -87,9 +87,8 @@ export class MachineSession {
// that frame can come after the row's Enter (replayed after a reconnect, or handled late): the
// PC's clock now would bound it too tightly, but the PC stamped the row before typing it, which
// bounds it though the chat never read the pane. An older bridge does not: only the turns' times do then
const after = waitingBaseline(key), stamped = item.sending_at === undefined ? Number.NaN : Date.parse(item.sending_at);
const pcFloor = Number.isFinite(stamped) ? stamped : null;
this.going.set(item.id, { at: Date.now(), after: after !== null ? { ...after, pcFloor } : pcFloor === null ? null : { keys: new Set(), latest: null, takenAt: Date.now(), pcFloor } });
const after = waitingBaseline(key);
this.going.set(item.id, { at: Date.now(), after: stampedBaseline(after !== null ? { ...after, pcFloor: null } : null, item.sending_at) });
}
else if (item.state === "queued" || item.state === "held") this.going.delete(item.id);
}
Expand Down
34 changes: 33 additions & 1 deletion src/lib/sentWaiting.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { describe, expect, it } from "bun:test";

import { clearMachineWaitingSends, currentWaitingSends, expireWaitingSends, forgetMachineWaitingSends, forgetWaitingBaseline, forgetWaitingSends, observeWaitingClock, pruneWaitingSends, rememberWaitingSend, settleWaitingSends, WAITING_SEND_MS, waitingBaseline, waitingKey } from "./sentWaiting.ts";
import { clearMachineWaitingSends, currentWaitingSends, expireWaitingSends, forgetMachineWaitingSends, forgetWaitingBaseline, forgetWaitingSends, observeWaitingClock, pruneWaitingSends, rememberWaitingSend, settleWaitingSends, stampedBaseline, WAITING_SEND_MS, waitingBaseline, waitingKey } from "./sentWaiting.ts";
import type { ConversationTurn } from "../../shared/protocol.ts";
import type { Machine } from "../../shared/machines.ts";
import { paneLifetimes } from "./paneLifetime.ts";
Expand Down Expand Up @@ -254,6 +254,38 @@ describe("waitingText", () => {
} finally { forgetWaitingSends(key); }
});

it("settles an immediate send by the PC's stamp though the PC's clock was set back since the chat's last read", () => {
const key = waitingKey("local", "pc-set-back");
const pc = Date.parse("2026-11-05T10:00:00Z");
try {
const earlier = user("earlier", pc + 37);
settleWaitingSends(key, [earlier], pc + 10_100, true);
observeWaitingClock(key, new Date(pc + 10_037).toISOString(), 100);
// the PC's clock was set back 30 s; 2 s after the read the send goes, and the PC stamps it before typing
const sent = stampedBaseline(waitingBaseline(key, 2_100), new Date(pc - 17_963).toISOString(), pc + 12_100);
rememberWaitingSend(key, "after the correction", pc + 12_100, sent);
settleWaitingSends(key, [earlier, user("after the correction", pc - 17_800)], pc + 12_400, true);
expect(currentWaitingSends(key)).toEqual([]);
} finally { forgetWaitingSends(key); }
});

it("keeps an immediate send past an identical turn written just before it an hour after the chat's last read", () => {
const key = waitingKey("local", "stamped-late");
const pc = Date.parse("2026-11-05T10:00:00Z");
try {
const earlier = user("earlier", pc + 37);
settleWaitingSends(key, [earlier], pc + 600_000, true);
observeWaitingClock(key, new Date(pc + 600_000).toISOString(), 0);
// a hidden tab sends an hour later; the terminal wrote the same words a second before the PC stamped the send
const sent = stampedBaseline(waitingBaseline(key, 3_600_000), new Date(pc + 4_199_037).toISOString(), pc + 4_200_000);
rememberWaitingSend(key, "repeat", pc + 4_200_000, sent);
settleWaitingSends(key, [earlier, user("repeat", pc + 4_198_037)], pc + 4_200_300, true);
expect(currentWaitingSends(key).map((send) => send.text)).toEqual(["repeat"]);
settleWaitingSends(key, [earlier, user("repeat", pc + 4_198_037), user("repeat", pc + 4_199_237)], pc + 4_200_500, true);
expect(currentWaitingSends(key)).toEqual([]);
} finally { forgetWaitingSends(key); }
});

it("settles a send by its own turn though this browser's wall clock stepped since the chat's read", () => {
const key = waitingKey("local", "clock-step");
const pc = Date.parse("2026-11-05T10:00:00Z");
Expand Down
15 changes: 14 additions & 1 deletion src/lib/sentWaiting.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ export interface WaitingBaseline {
takenAt: number;
/**
* The PC's last answer to this chat's reads and what this browser's monotonic clock says has passed
* since (observeWaitingClock), or the time the PC stamped on a queued row as it went out. null from
* since (observeWaitingClock), or the time the PC stamped on a send before typing it (stampedBaseline). null from
* an older bridge, which says neither: the turns' times alone bound the skew.
*/
pcFloor: number | null;
Expand Down Expand Up @@ -175,6 +175,19 @@ export function observeWaitingClock(key: string, serverTime: string | null | und
if (Number.isFinite(pcTime) && !((clocks.get(key)?.received ?? Number.NEGATIVE_INFINITY) > received)) clocks.set(key, { pcTime, received, receivedWall });
}

/**
* `after` bounded by the PC's own stamp of when it began typing the send (a queued row's
* `sending_at`, or a submit-result's): its turn is written after it on the PC's clock alone, whatever
* that clock or this browser's did since the chat's last read. A shown turn's time does not bound it
* then: a clock set back since writes the send's turn before it. Without a stamp (an older bridge)
* `after` stays as it is; `takenAt` (the browser's clock) dates a baseline made from the stamp alone.
*/
export function stampedBaseline(after: WaitingBaseline | null, sendingAt: string | undefined, takenAt = Date.now()): WaitingBaseline | null {
const stamped = sendingAt === undefined ? Number.NaN : Date.parse(sendingAt);
if (!Number.isFinite(stamped)) return after;
return { keys: after?.keys ?? new Set(), latest: null, takenAt: after?.takenAt ?? takenAt, pcFloor: stamped };
}

export function rememberWaitingSend(key: string, text: string, at = Date.now(), after: WaitingBaseline | null = null): void {
// the text the composer types: paste markers it strips are no part of the turn
const typed = composerMessage(text);
Expand Down
Loading
Loading