Skip to content

feat: tolerate sealed session-history shapes in all readers (phase 2a) #359

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zxch3n wants to merge 3 commits into from
Open

feat: tolerate sealed session-history shapes in all readers (phase 2a) #359

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
90d0018
fix(history-import): align open-conversation bench with the pinned ti…
zxch3n Sep 3, 2026
e4b8e0b
feat: tolerate sealed session-history shapes in all readers (phase 2a)
zxch3n Sep 3, 2026
b387ec8
fix: pair imported history hashes with their cursor version
zxch3n Sep 6, 2026
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
23 changes: 23 additions & 0 deletions apps/cli/src/commands/session.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
type LocalProjectGitState,
type MachineId,
type MachineMeta,
type MessageContent,
type SessionHistoryInput,
type SessionId,
type SessionMeta,
Expand Down Expand Up @@ -1400,6 +1401,16 @@ describe('session command helpers', () => {
});

it('builds transcript entries from visible user and assistant content only', () => {
// Sealed turns persist tool_call skeletons (no toolCallId/content); the
// transcript path shared by `session history` and MCP `lody_session_history`
// must skip them like any other tool call.
const skeleton = {
type: 'tool_call',
kind: 'execute',
status: 'completed',
title: 'Shell: ls',
ref: { machineId: 'machine-id', turnId: 'assistant-entry', index: 1 },
} as unknown as MessageContent;
expect(
toSessionTranscriptEntries([
createHistoryEntry({
Expand Down Expand Up @@ -1428,6 +1439,7 @@ describe('session command helpers', () => {
{ type: 'thought', text: 'internal reasoning' },
{ type: 'tool_call', toolCallId: 'tool-1', status: 'completed' },
{ type: 'text', text: 'Final answer' },
skeleton,
{ type: 'text', text: 'Second paragraph' },
{ type: 'tool_call', toolCallId: 'tool-2', status: 'completed' },
],
Expand Down Expand Up @@ -1512,6 +1524,17 @@ describe('session command helpers', () => {
expect(renderAssistantTurnCompletion([{ type: 'tool_call', toolCallId: 'tool-1' }])).toBe(
'No visible assistant reply found.'
);
// A sealed tool_call skeleton (no toolCallId/content) is likewise invisible.
expect(
renderAssistantTurnCompletion([
{
type: 'tool_call',
kind: 'execute',
status: 'completed',
ref: { machineId: 'machine-id', turnId: 'turn-1', index: 0 },
} as unknown as MessageContent,
])
).toBe('No visible assistant reply found.');
});

it('waits only when --wait is explicit, independently of JSON output', () => {
Expand Down
94 changes: 75 additions & 19 deletions apps/cli/src/lib/local-project-history-sync-service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import {
isActiveSessionStatus,
SessionStatusFactory,
type ProjectRef,
type SessionHistoryInput,
type SessionId,
} from '@lody/shared';

Expand All @@ -35,9 +36,10 @@ import {
getHistoryImportKey,
getProviderLabel,
hasPendingDispatchHistory,
hashHistoryEntry,
hashHistoryEntryForVersion,
materializeReplay,
resolveImportedTurnHashes,
resolveStoredHashVersion,
resolveSessionTitle,
resolveSourceUpdatedAtMs,
selectLatestCatalogItems,
Expand Down Expand Up @@ -111,24 +113,47 @@ function emptySummary(): LocalProjectHistorySyncSummary {
async function readSessionImportedTurnHashes(
sessionDoc: SessionDocument,
externalHistory: ExternalAcpHistorySyncMeta
): Promise<readonly string[]> {
): Promise<{ importedTurnHashes: readonly string[]; importedTurnHashVersion: number }> {
const cursor = await sessionDoc.getExternalHistoryCursor();
return resolveImportedTurnHashes(externalHistory, cursor?.importedTurnHashes);
return {
importedTurnHashes: resolveImportedTurnHashes(externalHistory, cursor?.importedTurnHashes),
importedTurnHashVersion: resolveStoredHashVersion(
cursor?.importedTurnHashes !== undefined ? cursor : externalHistory
),
};
}

async function writeSessionImportedTurnHashes(
sessionDoc: SessionDocument,
turnHashes: readonly string[]
turnHashes: readonly string[],
hashVersion: number
): Promise<void> {
const current = await sessionDoc.getExternalHistoryCursor();
if (areStringArraysEqual(current?.importedTurnHashes ?? [], turnHashes)) {
if (
current?.hashVersion === hashVersion &&
areStringArraysEqual(current?.importedTurnHashes ?? [], turnHashes)
) {
return;
}
await sessionDoc.setExternalHistoryCursor({
importedTurnHashes: [...turnHashes],
hashVersion,
});
Comment on lines 138 to 141

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Keep cursor hashes paired with their version

If the cursor write succeeds but the following doc-meta update fails or the process exits, the session is left with v2 importedTurnHashes in the cursor and a v1 externalHistory.hashVersion. On the next source update, readSessionImportedTurnHashes returns those v2 hashes while decideHistoryRefresh interprets them using the meta's v1 version, so the valid imported prefix is reported as prefix_mismatch and the durable session is incorrectly marked sync_conflict. Read and compare the cursor's hashVersion together with its hashes, or otherwise make the version transition recoverable across the two non-atomic writes.

Useful? React with 👍 / 👎.

}

/**
* Hash locally stored turns in the version of the stored sync cursor. The
* decisions compare these against stored-version replay hashes, so hashing
* with the wrong canonical form (e.g. v2 against a v1 cursor) would
* manufacture a conflict on upgrade.
*/
function hashHistoryForStoredVersion(
history: readonly SessionHistoryInput[],
hashVersion: number
): string[] {
return history.map((entry) => hashHistoryEntryForVersion(entry, hashVersion));
}

async function listWorkspaceSessionMetas(
manager: LoroDocumentManager
): Promise<Array<{ sessionId: SessionId; meta: SessionMeta }>> {
Expand Down Expand Up @@ -413,12 +438,15 @@ export class LocalProjectHistorySyncService {
throw new Error('Imported session metadata no longer matches the selected ACP history.');
}
if (existingExternalHistory.status !== 'sync_conflict') {
const importedTurnHashes = await readSessionImportedTurnHashes(
const { importedTurnHashes, importedTurnHashVersion } = await readSessionImportedTurnHashes(
sessionDoc,
existingExternalHistory
);
if (
areStringArraysEqual(currentHistoryBeforeReplay.map(hashHistoryEntry), importedTurnHashes)
areStringArraysEqual(
hashHistoryForStoredVersion(currentHistoryBeforeReplay, importedTurnHashVersion),
importedTurnHashes
)
) {
return finishResolved(meta);
}
Expand Down Expand Up @@ -461,16 +489,20 @@ export class LocalProjectHistorySyncService {
throw new Error('Imported session metadata no longer matches the selected ACP history.');
}

const latestImportedTurnHashes = await readSessionImportedTurnHashes(
sessionDoc,
latestExternalHistory
);
const {
importedTurnHashes: latestImportedTurnHashes,
importedTurnHashVersion: latestImportedTurnHashVersion,
} = await readSessionImportedTurnHashes(sessionDoc, latestExternalHistory);
const latestHistory = await sessionDoc.getHistory();
const decision = decideHistoryConflictResolution({
externalHistory: latestExternalHistory,
importedTurnHashes: latestImportedTurnHashes,
importedTurnHashVersion: latestImportedTurnHashVersion,
materialized,
currentHistoryHashes: latestHistory.map(hashHistoryEntry),
currentHistoryHashes: hashHistoryForStoredVersion(
latestHistory,
latestImportedTurnHashVersion
),
currentHistoryHasPendingDispatch: hasPendingDispatchHistory(latestHistory),
});
if (decision.status === 'blocked') {
Expand All @@ -493,8 +525,9 @@ export class LocalProjectHistorySyncService {
const writeTimeDecision = decideHistoryConflictResolution({
externalHistory: latestExternalHistory,
importedTurnHashes: latestImportedTurnHashes,
importedTurnHashVersion: latestImportedTurnHashVersion,
materialized,
currentHistoryHashes: history.map(hashHistoryEntry),
currentHistoryHashes: hashHistoryForStoredVersion(history, latestImportedTurnHashVersion),
currentHistoryHasPendingDispatch: hasPendingDispatchHistory(history),
});
if (writeTimeDecision.status !== 'replace') {
Expand All @@ -506,7 +539,11 @@ export class LocalProjectHistorySyncService {
}
return materialized.history;
});
await writeSessionImportedTurnHashes(sessionDoc, materialized.turnHashes);
await writeSessionImportedTurnHashes(
sessionDoc,
materialized.turnHashes,
materialized.hashVersion
);
await this.manager.repo.upsertDocMeta(roomId, {
origin: 'external-acp',
lastMessageAt,
Expand Down Expand Up @@ -692,7 +729,11 @@ export class LocalProjectHistorySyncService {
try {
const sessionDoc = await this.manager.getOrCreateSessionDoc(sessionId);
await sessionDoc.updateHistory(() => args.materialized.history);
await writeSessionImportedTurnHashes(sessionDoc, args.materialized.turnHashes);
await writeSessionImportedTurnHashes(
sessionDoc,
args.materialized.turnHashes,
args.materialized.hashVersion
);
await this.manager.repo.upsertDocMeta(roomId, meta);
const synced = await sessionDoc.waitUntilSynced();
if (!synced) {
Expand Down Expand Up @@ -760,17 +801,26 @@ export class LocalProjectHistorySyncService {
nowIso: new Date(getServerNow()).toISOString(),
});
const sessionDoc = await this.manager.getOrCreateSessionDoc(args.existing.sessionId);
const importedTurnHashes = await readSessionImportedTurnHashes(sessionDoc, externalHistory);
const { importedTurnHashes, importedTurnHashVersion } = await readSessionImportedTurnHashes(
sessionDoc,
externalHistory
);

const replayDecision = decideHistoryRefresh({
externalHistory,
importedTurnHashes,
importedTurnHashVersion,
replayDigest: materialized.replayDigest,
turnHashes: materialized.turnHashes,
materialized,
});

if (replayDecision.reason === 'digest_match') {
await writeSessionImportedTurnHashes(sessionDoc, materialized.turnHashes);
await writeSessionImportedTurnHashes(
sessionDoc,
materialized.turnHashes,
materialized.hashVersion
);
await this.manager.repo.upsertDocMeta(getSessionRoomId(args.existing.sessionId), {
origin: 'external-acp',
externalHistory: buildExternalHistoryMeta({
Expand Down Expand Up @@ -798,9 +848,11 @@ export class LocalProjectHistorySyncService {
const appendDecision = decideHistoryRefresh({
externalHistory,
importedTurnHashes,
importedTurnHashVersion,
replayDigest: materialized.replayDigest,
turnHashes: materialized.turnHashes,
currentHistoryHashes: currentHistory.map(hashHistoryEntry),
materialized,
currentHistoryHashes: hashHistoryForStoredVersion(currentHistory, importedTurnHashVersion),
});
if (appendDecision.status === 'conflicted') {
await this.markConflict(
Expand All @@ -827,7 +879,11 @@ export class LocalProjectHistorySyncService {

const suffix = materialized.history.slice(appendDecision.appendFromIndex);
await sessionDoc.updateHistory((history) => [...history, ...suffix]);
await writeSessionImportedTurnHashes(sessionDoc, materialized.turnHashes);
await writeSessionImportedTurnHashes(
sessionDoc,
materialized.turnHashes,
materialized.hashVersion
);
await this.manager.repo.upsertDocMeta(getSessionRoomId(args.existing.sessionId), {
origin: 'external-acp',
lastMessageAt: resolveSourceUpdatedAtMs(args.info, getServerNow()),
Expand Down
Loading
Loading