agent-radar
7be535059e
Dependency audit / audit (push) Canceled after 0s
CI / plan (push) Canceled after 0s
Gitoxide helper admission / macos-latest (push) Canceled after 0s
Gitoxide helper admission / ubuntu-latest (push) Canceled after 0s
Gitoxide helper admission / windows-latest (push) Canceled after 0s
Runtime Host peer admission / quality (push) Canceled after 0s
Windows recovery / windows_recovery (push) Canceled after 0s
CI / heavy (push) Canceled after 0s
CI / test (push) Canceled after 0s
1023 lines
37 KiB
TypeScript
1023 lines
37 KiB
TypeScript
/*
|
|
* Licensed to the Apache Software Foundation (ASF) under one
|
|
* or more contributor license agreements. See the NOTICE file
|
|
* distributed with this work for additional information
|
|
* regarding copyright ownership. The ASF licenses this file
|
|
* to you under the Apache License, Version 2.0 (the
|
|
* "License"); you may not use this file except in compliance
|
|
* with the License. You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing,
|
|
* software distributed under the License is distributed on an
|
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
* KIND, either express or implied. See the License for the
|
|
* specific language governing permissions and limitations
|
|
* under the License.
|
|
*/
|
|
|
|
import { createHash, randomUUID } from 'node:crypto';
|
|
import { isDeepResearchSession } from '@maka/core/deep-research';
|
|
import { SIDE_CONVERSATION_SESSION_LABEL } from '@maka/core/side-conversation';
|
|
import type { RuntimeEvent } from '@maka/core/runtime-event';
|
|
import type { CreateSessionInput } from '@maka/core/runtime-inputs';
|
|
import {
|
|
isWorkHubCoordinationSessionId,
|
|
isWorkHubCoordinationSessionTarget,
|
|
sessionRevisionFamilyId,
|
|
type SessionConversationCopy,
|
|
type SessionHeader,
|
|
type StoredMessage,
|
|
} from '@maka/core/session';
|
|
import {
|
|
archivedToolResultContainsLinkedChildReferences,
|
|
archivedToolResultContainsConversationOwnedReferences,
|
|
cloneConversationRuntimeLedger,
|
|
collectConversationCopyLinkedChildReferences,
|
|
collectConversationCopySessionFileRefs,
|
|
createConversationCopySlice,
|
|
prepareConversationRuntimeLedgerCopy,
|
|
type ConversationRuntimeLedgerCopyPlan,
|
|
} from '@maka/runtime/conversation-copy';
|
|
import { isArchivedToolResultPlaceholder } from '@maka/runtime/context-budget';
|
|
import type { AgentRunEvent } from '@maka/core/agent-run';
|
|
import {
|
|
decodeModelProjectionTransition,
|
|
MODEL_PROJECTION_TRANSITION_EVENT_TYPE,
|
|
type ModelProjectionTransition,
|
|
} from '@maka/core/model-projection-transition';
|
|
import { type SessionManager } from '@maka/runtime/session-manager';
|
|
import {
|
|
authenticateInteractiveArtifactStoreWriter,
|
|
type InteractiveArtifactStoreWriter,
|
|
} from '@maka/storage/artifact-stores';
|
|
import {
|
|
authenticateExecutionStoresWriter,
|
|
isSessionNotFoundError,
|
|
type ExecutionStoresWriter,
|
|
} from '@maka/storage/execution-stores';
|
|
import {
|
|
authenticateInteractiveSessionTodoWriter,
|
|
type InteractiveSessionTodoWriter,
|
|
} from '@maka/storage/session-todo-authority';
|
|
import type {
|
|
OperationOutcome,
|
|
SessionConversationCopyInput,
|
|
SessionConversationCopyResult,
|
|
SessionRevisionAbandonInput,
|
|
SessionRevisionAbandonResult,
|
|
} from '../protocol/index.js';
|
|
import type {
|
|
SessionRevisionOperationHandlerMap,
|
|
SessionRevisionOperationKey,
|
|
} from './operation-dispatcher.js';
|
|
import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js';
|
|
import { projectSessionCatalogRecord } from './session-catalog-coordinator.js';
|
|
import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js';
|
|
import {
|
|
agentGraphRevisionAdmissionSessionIds,
|
|
prepareAgentGraphRevisionReferences,
|
|
} from './session-revision-graph-references.js';
|
|
import {
|
|
conversationCopyCommitFailureDiagnostic,
|
|
conversationCopyCommitFailureMessage,
|
|
} from './session-revision-diagnostics.js';
|
|
import { purgeSessionSidecars } from './session-sidecar-purge.js';
|
|
|
|
type ConversationCopyKind = 'branch' | 'revision';
|
|
type ConversationCopySemanticKind = ConversationCopyKind | 'side_conversation';
|
|
type ConversationCopyOperationKey = Exclude<
|
|
SessionRevisionOperationKey,
|
|
'session.revision.abandon'
|
|
>;
|
|
type ConversationCopyOutcome = OperationOutcome<ConversationCopyOperationKey>;
|
|
type RevisionAbandonOutcome = OperationOutcome<'session.revision.abandon'>;
|
|
const CONVERSATION_COPY_ADMISSION_PASSES = 2;
|
|
interface ConversationCopyAdmissionRetry {
|
|
readonly kind: 'retry_admission';
|
|
readonly sessionIds: readonly string[];
|
|
}
|
|
type ConversationCopyCreateInput = CreateSessionInput & {
|
|
readonly conversationCopy: SessionConversationCopy;
|
|
};
|
|
|
|
export interface HostSessionRevisionCoordinatorOptions {
|
|
readonly stores: ExecutionStoresWriter<'interactive'>;
|
|
readonly artifacts: InteractiveArtifactStoreWriter;
|
|
readonly sessionTodo: InteractiveSessionTodoWriter;
|
|
readonly manager: SessionManager;
|
|
readonly admission: SessionAdmissionGate;
|
|
readonly continuity: SessionContinuityCoordinator;
|
|
readonly graph: Pick<
|
|
import('@maka/runtime/stream-graph-coordinator').AgentGraphCoordinator,
|
|
'readGraphState' | 'readSessionState'
|
|
>;
|
|
readonly isSessionActive: (sessionId: string) => boolean;
|
|
readonly requestDrain: () => void;
|
|
}
|
|
|
|
/** Host authority for exact, retryable cross-Session branch and revision copies. */
|
|
export class HostSessionRevisionCoordinator {
|
|
readonly handlers: SessionRevisionOperationHandlerMap = {
|
|
'session.branch.create': (input) => this.#copy('branch', input),
|
|
'session.revision.create': (input) => this.#copy('revision', input),
|
|
'session.revision.abandon': (input) => this.#abandonRevision(input),
|
|
};
|
|
|
|
readonly #stores: ExecutionStoresWriter<'interactive'>;
|
|
readonly #artifacts: InteractiveArtifactStoreWriter;
|
|
readonly #sessionTodo: InteractiveSessionTodoWriter;
|
|
|
|
constructor(private readonly options: HostSessionRevisionCoordinatorOptions) {
|
|
this.#stores = authenticateExecutionStoresWriter(options.stores, 'interactive');
|
|
this.#artifacts = authenticateInteractiveArtifactStoreWriter(options.artifacts);
|
|
this.#sessionTodo = authenticateInteractiveSessionTodoWriter(options.sessionTodo);
|
|
}
|
|
|
|
async recover(): Promise<void> {
|
|
const copies = (await this.#stores.sessionStore.listHeaders()).filter(
|
|
(header) => header.conversationCopy !== undefined,
|
|
);
|
|
for (const header of copies) {
|
|
if (header.conversationCopy!.state === 'preparing') await this.#discard(header);
|
|
}
|
|
|
|
const committed = copies.filter((header) => header.conversationCopy!.state === 'committed');
|
|
const retained = new Set(
|
|
committed
|
|
.filter(
|
|
(header) =>
|
|
header.conversationCopy!.kind === 'branch' || header.revisionState === 'committed',
|
|
)
|
|
.map((header) => header.id),
|
|
);
|
|
for (const header of committed) {
|
|
if (
|
|
header.conversationCopy!.kind === 'revision' &&
|
|
header.revisionState === 'preparing' &&
|
|
(await this.#hasAdmittedRevisionTurn(header.id))
|
|
) {
|
|
retained.add(header.id);
|
|
}
|
|
}
|
|
for (let changed = true; changed; ) {
|
|
changed = false;
|
|
for (const header of committed) {
|
|
if (!retained.has(header.id)) continue;
|
|
const sourceSessionId = header.conversationCopy!.sourceSessionId;
|
|
if (!retained.has(sourceSessionId)) {
|
|
retained.add(sourceSessionId);
|
|
changed = true;
|
|
}
|
|
}
|
|
}
|
|
for (const header of committed) {
|
|
if (header.conversationCopy!.kind !== 'revision' || header.revisionState !== 'preparing') {
|
|
continue;
|
|
}
|
|
if (retained.has(header.id)) {
|
|
await this.options.manager.commitRevisionVersion(header.id);
|
|
} else {
|
|
await this.#discard(header);
|
|
}
|
|
}
|
|
}
|
|
|
|
async #copy(
|
|
kind: ConversationCopyKind,
|
|
input: SessionConversationCopyInput,
|
|
): Promise<ConversationCopyOutcome> {
|
|
if (isWorkHubCoordinationSessionId(input.targetSessionId)) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Target Session identity is reserved for WorkHub coordination',
|
|
);
|
|
}
|
|
if (isWorkHubCoordinationSessionId(input.sourceSessionId)) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'WorkHub Coordination Session cannot be copied as an ordinary conversation',
|
|
);
|
|
}
|
|
const semanticKind = conversationCopySemanticKind(kind, input);
|
|
const requestFingerprint = conversationCopyFingerprint(semanticKind, input);
|
|
const retry = await this.options.admission.run(input.targetSessionId, async () =>
|
|
this.#resolveExistingTarget(semanticKind, input, requestFingerprint, true),
|
|
);
|
|
if (retry) return retry;
|
|
|
|
let rootSessionId: string;
|
|
try {
|
|
const source = await this.#stores.sessionStore.readHeaderRecordSnapshot(
|
|
input.sourceSessionId,
|
|
);
|
|
rootSessionId = source.header.revisionRootSessionId ?? input.sourceSessionId;
|
|
} catch (error) {
|
|
return isSessionNotFoundError(error)
|
|
? copyFailure('not_found', 'Source Session does not exist')
|
|
: copyFailure('persistence_failed', 'Source Session metadata is unavailable');
|
|
}
|
|
|
|
const admittedSessionIds = new Set([
|
|
input.sourceSessionId,
|
|
input.targetSessionId,
|
|
rootSessionId,
|
|
]);
|
|
// The first admitted read discovers only children retained by this copy.
|
|
// Artifact deletion uses each child Session lane, so retry with those exact
|
|
// lanes and keep the complete lease through validation and publication.
|
|
for (let pass = 0; pass < CONVERSATION_COPY_ADMISSION_PASSES; pass += 1) {
|
|
const result = await this.options.admission.runMany([...admittedSessionIds], (lease) =>
|
|
this.#copyAdmitted(semanticKind, input, requestFingerprint, lease, admittedSessionIds),
|
|
);
|
|
if ('ok' in result) return result;
|
|
for (const sessionId of result.sessionIds) admittedSessionIds.add(sessionId);
|
|
}
|
|
return copyFailure('session_busy', 'Agent Graph child ownership changed during copy');
|
|
}
|
|
|
|
async #abandonRevision(input: SessionRevisionAbandonInput): Promise<RevisionAbandonOutcome> {
|
|
return this.options.admission.run(input.targetSessionId, async () => {
|
|
let header: SessionHeader;
|
|
try {
|
|
header = await this.#stores.sessionStore.readHeaderSnapshot(input.targetSessionId);
|
|
} catch (error) {
|
|
return isSessionNotFoundError(error)
|
|
? abandonFailure('not_found', 'Target Session revision does not exist')
|
|
: abandonFailure('persistence_failed', 'Target Session revision is unavailable');
|
|
}
|
|
if (header.conversationCopy?.kind !== 'revision') {
|
|
return abandonFailure('operation_conflict', 'Target is not a Session revision copy');
|
|
}
|
|
if (
|
|
header.revisionState !== 'preparing' ||
|
|
this.options.isSessionActive(input.targetSessionId) ||
|
|
(await this.#hasAdmittedRevisionTurn(input.targetSessionId)) ||
|
|
(await this.#hasCommittedConversationCopyDependent(input.targetSessionId))
|
|
) {
|
|
return abandonSuccess({ kind: 'retained', sessionId: input.targetSessionId });
|
|
}
|
|
try {
|
|
await this.#discard(header);
|
|
return abandonSuccess({
|
|
kind: 'abandoned',
|
|
sessionId: input.targetSessionId,
|
|
});
|
|
} catch {
|
|
return abandonFailure(
|
|
'persistence_failed',
|
|
'Target Session revision could not be abandoned',
|
|
);
|
|
}
|
|
});
|
|
}
|
|
|
|
async #copyAdmitted(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
requestFingerprint: `sha256:${string}`,
|
|
lease: SessionAdmissionLease,
|
|
admittedSessionIds: ReadonlySet<string>,
|
|
): Promise<ConversationCopyOutcome | ConversationCopyAdmissionRetry> {
|
|
const existing = await this.#resolveExistingTarget(kind, input, requestFingerprint, false);
|
|
if (existing) return existing;
|
|
|
|
let sourceRecord;
|
|
try {
|
|
sourceRecord = await this.#stores.sessionStore.readHeaderRecordSnapshot(
|
|
input.sourceSessionId,
|
|
);
|
|
} catch (error) {
|
|
return isSessionNotFoundError(error)
|
|
? copyFailure('not_found', 'Source Session does not exist')
|
|
: copyFailure('persistence_failed', 'Source Session metadata is unavailable');
|
|
}
|
|
if (sourceRecord.revision !== input.expectedSourceRevision) {
|
|
return copySuccess({
|
|
kind: 'source_revision_conflict',
|
|
expectedRevision: input.expectedSourceRevision,
|
|
actualRevision: sourceRecord.revision,
|
|
});
|
|
}
|
|
const sourceHeader = sourceRecord.header;
|
|
if (sourceHeader.conversationCopy?.state === 'preparing') {
|
|
return copyFailure('not_found', 'Source Session does not exist');
|
|
}
|
|
if (kind === 'revision' && sourceHeader.isArchived) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Archived Session revision families cannot create active revisions',
|
|
);
|
|
}
|
|
if (isWorkHubCoordinationSessionTarget(sourceHeader)) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'WorkHub Coordination Session cannot be copied as an ordinary conversation',
|
|
);
|
|
}
|
|
if (sourceHeader.subagentParent) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Linked child Sessions cannot be copied as ordinary conversations',
|
|
);
|
|
}
|
|
if (isDeepResearchSession(sourceHeader.labels)) {
|
|
return copyFailure(
|
|
'operation_unavailable',
|
|
'Deep Research Sessions cannot be copied without an exact research ledger boundary',
|
|
);
|
|
}
|
|
if (kind !== 'side_conversation' && this.options.isSessionActive(input.sourceSessionId)) {
|
|
return copyFailure('session_busy', 'Source Session has an active Turn');
|
|
}
|
|
|
|
let source;
|
|
try {
|
|
source = await this.options.manager.readConversationCopySnapshot(input.sourceSessionId);
|
|
} catch {
|
|
return copyFailure('persistence_failed', 'Source conversation ledger is unavailable');
|
|
}
|
|
const slice = createConversationCopySlice(
|
|
source.messages,
|
|
input.sourceTurnId,
|
|
kind === 'revision' ? 'before' : 'through',
|
|
);
|
|
if (!slice) {
|
|
return copyFailure('invalid_request', 'Source turn does not exist');
|
|
}
|
|
let plan: ConversationRuntimeLedgerCopyPlan;
|
|
let sessionHeaders: SessionHeader[];
|
|
try {
|
|
[plan, sessionHeaders] = await Promise.all([
|
|
prepareConversationRuntimeLedgerCopy({
|
|
sourceSessionId: input.sourceSessionId,
|
|
sourceEvents: source.events,
|
|
copiedMessages: slice.messages,
|
|
runStore: this.#stores.agentRunStore,
|
|
runtimeEventStore: this.#stores.runtimeEventStore,
|
|
}),
|
|
this.#stores.sessionStore.listHeaders(),
|
|
]);
|
|
} catch (error) {
|
|
if (isConversationRuntimeFactRewriteUnsupported(error)) {
|
|
return copyFailure(
|
|
'operation_unavailable',
|
|
'Session conversation copy does not yet support continuation authority facts',
|
|
);
|
|
}
|
|
return copyFailure('persistence_failed', 'Source conversation lineage is unavailable');
|
|
}
|
|
if (
|
|
kind === 'revision' &&
|
|
sessionHeaders.some(
|
|
(candidate) =>
|
|
sessionRevisionFamilyId(candidate) === sessionRevisionFamilyId(sourceHeader) &&
|
|
candidate.isArchived,
|
|
)
|
|
) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Archived Session revision families cannot create active revisions',
|
|
);
|
|
}
|
|
const copyTurnIds = plan.copyTurnIds;
|
|
const archivePreflight = await this.#readArchivedToolResults(
|
|
input.sourceSessionId,
|
|
plan.runs.flatMap(({ runtimeEvents }) => runtimeEvents),
|
|
slice.messages,
|
|
copyTurnIds,
|
|
plan.runs.flatMap(({ operationalEvents }) => operationalEvents),
|
|
);
|
|
if (!archivePreflight.ok) return archivePreflight.outcome;
|
|
const linkedChildRequests = collectConversationCopyLinkedChildReferences({
|
|
messages: slice.messages,
|
|
runtimeEvents: plan.runs.flatMap(({ runtimeEvents }) => runtimeEvents),
|
|
archivedResults: archivePreflight.serializedResults,
|
|
});
|
|
const referencedSessionFileIds = collectConversationCopySessionFileRefs({
|
|
sourceSessionId: input.sourceSessionId,
|
|
messages: slice.messages,
|
|
runtimeEvents: plan.runs.flatMap(({ runtimeEvents }) => runtimeEvents),
|
|
archivedResults: archivePreflight.serializedResults,
|
|
});
|
|
const missingGraphChildSessionIds = agentGraphRevisionAdmissionSessionIds({
|
|
sourceSessionId: input.sourceSessionId,
|
|
sessionHeaders,
|
|
copyTurnIds,
|
|
requests: linkedChildRequests,
|
|
}).filter((sessionId) => !admittedSessionIds.has(sessionId));
|
|
if (missingGraphChildSessionIds.length > 0) {
|
|
return { kind: 'retry_admission', sessionIds: missingGraphChildSessionIds };
|
|
}
|
|
const linkedReferences = await prepareAgentGraphRevisionReferences(
|
|
{
|
|
kind,
|
|
sourceSessionId: input.sourceSessionId,
|
|
sourceHeader,
|
|
sessionHeaders,
|
|
copyTurnIds,
|
|
requests: linkedChildRequests,
|
|
},
|
|
{
|
|
agentRunStore: this.#stores.agentRunStore,
|
|
artifacts: this.#artifacts,
|
|
graph: this.options.graph,
|
|
isSessionActive: this.options.isSessionActive,
|
|
},
|
|
);
|
|
if (!linkedReferences.ok) {
|
|
return copyFailure(linkedReferences.code, linkedReferences.message);
|
|
}
|
|
if (
|
|
kind !== 'side_conversation' &&
|
|
archivePreflight.serializedResults.some((serializedResult) =>
|
|
archivedToolResultContainsConversationOwnedReferences(
|
|
serializedResult,
|
|
input.sourceSessionId,
|
|
linkedReferences.references,
|
|
),
|
|
)
|
|
) {
|
|
return copyFailure(
|
|
'operation_unavailable',
|
|
'Session conversation copy cannot preserve owned references inside archived tool results',
|
|
);
|
|
}
|
|
|
|
let createInput: ConversationCopyCreateInput;
|
|
try {
|
|
createInput = await this.#createInput(kind, input, requestFingerprint, sourceHeader);
|
|
} catch {
|
|
return copyFailure('persistence_failed', 'Session revision family is unavailable');
|
|
}
|
|
let boundary;
|
|
try {
|
|
boundary = await this.#stores.sessionStore.readExecutionBoundary(input.sourceSessionId);
|
|
} catch {
|
|
return copyFailure('persistence_failed', 'Source execution boundary is unavailable');
|
|
}
|
|
|
|
const created = await this.#stores.sessionStore
|
|
.createStableSession(
|
|
{
|
|
sessionId: input.targetSessionId,
|
|
requestFingerprint,
|
|
input: createInput,
|
|
},
|
|
boundary,
|
|
)
|
|
.catch(() => null);
|
|
if (!created) {
|
|
return this.#unknownAfterCommitAttempt(
|
|
kind,
|
|
input,
|
|
requestFingerprint,
|
|
'Session conversation-copy creation outcome is unknown',
|
|
);
|
|
}
|
|
if (created.kind === 'conflict') {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Target Session identity belongs to a different request',
|
|
);
|
|
}
|
|
if (created.kind === 'existing') {
|
|
return (
|
|
(await this.#resolveExistingTarget(kind, input, requestFingerprint, false)) ??
|
|
copyFailure('commit_outcome_unknown', 'Target Session publication state is unknown')
|
|
);
|
|
}
|
|
|
|
try {
|
|
const archivedSnapshotResults = new Map(
|
|
archivePreflight.results
|
|
.filter(
|
|
({ serializedResult }) =>
|
|
archivedToolResultContainsLinkedChildReferences(serializedResult) ||
|
|
archivedToolResultContainsConversationOwnedReferences(
|
|
serializedResult,
|
|
input.sourceSessionId,
|
|
linkedReferences.references,
|
|
),
|
|
)
|
|
.map(({ descriptor, serializedResult }) => [descriptor.artifactId, serializedResult]),
|
|
);
|
|
const artifactCopy = await this.#artifacts.copyConversationArtifacts({
|
|
sourceSessionId: input.sourceSessionId,
|
|
targetSessionId: input.targetSessionId,
|
|
turnIds: copyTurnIds,
|
|
...(referencedSessionFileIds.size > 0
|
|
? { includeArtifactIds: [...referencedSessionFileIds] }
|
|
: {}),
|
|
...(kind === 'side_conversation' && archivedSnapshotResults.size > 0
|
|
? { excludeArtifactIds: [...archivedSnapshotResults.keys()] }
|
|
: {}),
|
|
...(kind === 'side_conversation' && linkedReferences.references.size > 0
|
|
? {
|
|
linkedArtifacts: [...linkedReferences.references].map(([sessionId, references]) => ({
|
|
sessionId,
|
|
artifactIds: [...references.artifactIds],
|
|
})),
|
|
}
|
|
: {}),
|
|
});
|
|
const references = {
|
|
mode: 'exact' as const,
|
|
sourceSessionId: input.sourceSessionId,
|
|
targetSessionId: input.targetSessionId,
|
|
artifactIds: artifactCopy.artifactIds,
|
|
relativePaths: artifactCopy.relativePaths,
|
|
linkedChildren:
|
|
kind === 'side_conversation'
|
|
? {
|
|
mode: 'snapshot' as const,
|
|
archivedResults: archivedSnapshotResults,
|
|
}
|
|
: linkedReferences.references.size > 0
|
|
? {
|
|
mode: 'preserve_validated' as const,
|
|
references: linkedReferences.references,
|
|
}
|
|
: { mode: 'reject' as const },
|
|
};
|
|
const runtimeCopy = await cloneConversationRuntimeLedger({
|
|
plan,
|
|
copiedMessages: slice.messages,
|
|
referenceMap: references,
|
|
runStore: this.#stores.agentRunStore,
|
|
runtimeEventStore: this.#stores.runtimeEventStore,
|
|
newId: randomUUID,
|
|
});
|
|
const copiedMessages = runtimeCopy.copiedMessages;
|
|
await this.#sessionTodo.initializeCopy({
|
|
sourceSessionId: input.sourceSessionId,
|
|
targetSessionId: input.targetSessionId,
|
|
copyCurrent: kind === 'branch' && slice.beforeTs === undefined,
|
|
});
|
|
if (copiedMessages.length > 0) {
|
|
await this.#stores.sessionStore.appendMessages(input.targetSessionId, [...copiedMessages]);
|
|
}
|
|
await this.#stores.sessionStore.appendMessage(
|
|
input.targetSessionId,
|
|
conversationCopyStartNote(kind, input, createInput),
|
|
);
|
|
await this.#stores.sessionStore.updateHeader(input.targetSessionId, {
|
|
conversationCopy: {
|
|
...createInput.conversationCopy!,
|
|
state: 'committed',
|
|
},
|
|
isFlagged: sourceHeader.isFlagged,
|
|
titleIsManual: sourceHeader.titleIsManual,
|
|
connectionLocked:
|
|
sourceHeader.connectionLocked ||
|
|
copiedMessages.some((message) => message.type === 'user'),
|
|
});
|
|
await this.options.continuity.refreshCanonical(input.targetSessionId, lease);
|
|
return copySuccess({
|
|
kind: 'committed',
|
|
session: projectSessionCatalogRecord(
|
|
await this.#stores.sessionStore.readCatalogRecord(input.targetSessionId),
|
|
),
|
|
});
|
|
} catch (error) {
|
|
console.error(
|
|
`[runtime-host] ${kind} conversation copy commit failed (${input.sourceSessionId} -> ${input.targetSessionId}): ${conversationCopyCommitFailureDiagnostic(error)}`,
|
|
);
|
|
return this.#rollbackIncompleteCopy(
|
|
kind,
|
|
input,
|
|
requestFingerprint,
|
|
conversationCopyCommitFailureMessage(error),
|
|
);
|
|
}
|
|
}
|
|
|
|
async #readArchivedToolResults(
|
|
sourceSessionId: string,
|
|
sourceEvents: readonly RuntimeEvent[],
|
|
copiedMessages: readonly StoredMessage[],
|
|
copyTurnIds: readonly string[],
|
|
operationalEvents: readonly AgentRunEvent[],
|
|
): Promise<
|
|
| {
|
|
readonly ok: true;
|
|
readonly results: readonly {
|
|
readonly descriptor: ArchivedToolResultCopyDescriptor;
|
|
readonly serializedResult: string;
|
|
}[];
|
|
readonly serializedResults: readonly string[];
|
|
}
|
|
| { readonly ok: false; readonly outcome: ConversationCopyOutcome }
|
|
> {
|
|
const archives = collectArchivedToolResultPlaceholders(
|
|
sourceEvents,
|
|
copiedMessages,
|
|
copyTurnIds,
|
|
operationalEvents,
|
|
);
|
|
if (!archives) {
|
|
return {
|
|
ok: false,
|
|
outcome: copyFailure('persistence_failed', 'Archived tool result metadata is invalid'),
|
|
};
|
|
}
|
|
const results: Array<{
|
|
descriptor: ArchivedToolResultCopyDescriptor;
|
|
serializedResult: string;
|
|
}> = [];
|
|
for (const archive of archives) {
|
|
const read = await this.#artifacts
|
|
.readTextInSession(sourceSessionId, archive.artifactId, {
|
|
maxBytes: archive.originalBytes,
|
|
})
|
|
.catch(() => null);
|
|
if (
|
|
!read?.ok ||
|
|
Buffer.byteLength(read.text, 'utf8') !== archive.originalBytes ||
|
|
createHash('sha256').update(read.text).digest('hex') !== archive.bodySha256
|
|
) {
|
|
return {
|
|
ok: false,
|
|
outcome: copyFailure(
|
|
'persistence_failed',
|
|
'Archived tool result is unavailable or corrupt',
|
|
),
|
|
};
|
|
}
|
|
results.push({ descriptor: archive, serializedResult: read.text });
|
|
}
|
|
return {
|
|
ok: true,
|
|
results,
|
|
serializedResults: results.map(({ serializedResult }) => serializedResult),
|
|
};
|
|
}
|
|
|
|
async #createInput(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
requestFingerprint: `sha256:${string}`,
|
|
source: SessionHeader,
|
|
): Promise<ConversationCopyCreateInput> {
|
|
const common: ConversationCopyCreateInput = {
|
|
cwd: source.cwd,
|
|
...(source.projectId !== undefined ? { projectId: source.projectId } : {}),
|
|
...(source.llmConnectionId === undefined ? {} : { llmConnectionId: source.llmConnectionId }),
|
|
llmConnectionSlug: source.llmConnectionSlug,
|
|
model: source.model,
|
|
...(source.thinkingLevel !== undefined ? { thinkingLevel: source.thinkingLevel } : {}),
|
|
permissionMode: source.permissionMode,
|
|
collaborationMode: source.collaborationMode ?? 'agent',
|
|
orchestrationMode: source.orchestrationMode ?? 'default',
|
|
name: source.name,
|
|
labels:
|
|
kind === 'side_conversation'
|
|
? [...new Set([...source.labels, SIDE_CONVERSATION_SESSION_LABEL])]
|
|
: [...source.labels],
|
|
conversationCopy: {
|
|
kind: persistedConversationCopyKind(kind),
|
|
sourceSessionId: input.sourceSessionId,
|
|
sourceTurnId: input.sourceTurnId,
|
|
requestFingerprint,
|
|
state: 'preparing',
|
|
...(kind === 'side_conversation' ? { intent: kind } : {}),
|
|
},
|
|
status: 'active',
|
|
};
|
|
if (kind !== 'revision') {
|
|
return {
|
|
...common,
|
|
parentSessionId: input.sourceSessionId,
|
|
branchOfTurnId: input.sourceTurnId,
|
|
};
|
|
}
|
|
const revisionRootSessionId = source.revisionRootSessionId ?? input.sourceSessionId;
|
|
const family = (await this.#stores.sessionStore.listHeaders()).filter(
|
|
(candidate) =>
|
|
candidate.conversationCopy?.state !== 'preparing' &&
|
|
(candidate.id === revisionRootSessionId ||
|
|
candidate.revisionRootSessionId === revisionRootSessionId),
|
|
);
|
|
const revisionIndex =
|
|
Math.max(1, ...family.map((candidate) => candidate.revisionIndex ?? 1)) + 1;
|
|
return {
|
|
...common,
|
|
...(source.parentSessionId ? { parentSessionId: source.parentSessionId } : {}),
|
|
...(source.branchOfTurnId ? { branchOfTurnId: source.branchOfTurnId } : {}),
|
|
revisionRootSessionId,
|
|
revisionParentSessionId: input.sourceSessionId,
|
|
revisionOfTurnId: input.sourceTurnId,
|
|
revisionIndex,
|
|
revisionState: 'preparing',
|
|
};
|
|
}
|
|
|
|
async #resolveExistingTarget(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
requestFingerprint: `sha256:${string}`,
|
|
discardPreparing: boolean,
|
|
): Promise<ConversationCopyOutcome | null> {
|
|
let probe;
|
|
try {
|
|
probe = await this.#stores.sessionStore.probeStableSessionCreate(
|
|
input.targetSessionId,
|
|
requestFingerprint,
|
|
);
|
|
} catch {
|
|
return copyFailure('persistence_failed', 'Target Session identity is unavailable');
|
|
}
|
|
if (probe.kind === 'absent') return null;
|
|
if (probe.kind === 'conflict') {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Target Session identity belongs to a different request',
|
|
);
|
|
}
|
|
const copy = probe.record.header.conversationCopy;
|
|
if (
|
|
copy?.kind !== persistedConversationCopyKind(kind) ||
|
|
copy.intent !== (kind === 'side_conversation' ? kind : undefined) ||
|
|
copy.sourceSessionId !== input.sourceSessionId ||
|
|
copy.sourceTurnId !== input.sourceTurnId ||
|
|
copy.requestFingerprint !== requestFingerprint
|
|
) {
|
|
return copyFailure(
|
|
'operation_conflict',
|
|
'Target Session identity belongs to a different conversation copy',
|
|
);
|
|
}
|
|
if (copy.state === 'committed') {
|
|
try {
|
|
return copySuccess({
|
|
kind: 'committed',
|
|
session: projectSessionCatalogRecord(
|
|
await this.#stores.sessionStore.readCatalogRecord(input.targetSessionId),
|
|
),
|
|
});
|
|
} catch {
|
|
return copyFailure(
|
|
'commit_outcome_unknown',
|
|
'Committed target Session projection is unavailable',
|
|
);
|
|
}
|
|
}
|
|
if (!discardPreparing) return null;
|
|
try {
|
|
await this.#discard(probe.record.header);
|
|
return null;
|
|
} catch {
|
|
this.options.requestDrain();
|
|
return copyFailure(
|
|
'commit_outcome_unknown',
|
|
'Incomplete target Session could not be recovered',
|
|
);
|
|
}
|
|
}
|
|
|
|
async #rollbackIncompleteCopy(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
requestFingerprint: `sha256:${string}`,
|
|
message: string,
|
|
): Promise<ConversationCopyOutcome> {
|
|
let header;
|
|
try {
|
|
header = await this.#stores.sessionStore.readHeaderSnapshot(input.targetSessionId);
|
|
if (header.conversationCopy?.state === 'committed') {
|
|
return copySuccess({
|
|
kind: 'committed',
|
|
session: projectSessionCatalogRecord(
|
|
await this.#stores.sessionStore.readCatalogRecord(input.targetSessionId),
|
|
),
|
|
});
|
|
}
|
|
await this.#discard(header);
|
|
return copyFailure('persistence_failed', message);
|
|
} catch {
|
|
return this.#unknownAfterCommitAttempt(kind, input, requestFingerprint, message);
|
|
}
|
|
}
|
|
|
|
async #unknownAfterCommitAttempt(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
requestFingerprint: `sha256:${string}`,
|
|
message: string,
|
|
): Promise<ConversationCopyOutcome> {
|
|
const resolved = await this.#resolveExistingTarget(kind, input, requestFingerprint, false);
|
|
if (resolved?.ok) return resolved;
|
|
this.options.requestDrain();
|
|
return copyFailure('commit_outcome_unknown', message);
|
|
}
|
|
|
|
async #discard(header: SessionHeader): Promise<void> {
|
|
const copy = header.conversationCopy;
|
|
if (!copy) throw new Error('Session is not a conversation copy');
|
|
await purgeSessionSidecars(
|
|
{
|
|
artifacts: this.#artifacts,
|
|
sessionTodo: this.#sessionTodo,
|
|
purgeOperationalState: (sessionId) =>
|
|
this.#stores.purgeConversationOperationalState(sessionId),
|
|
},
|
|
header.id,
|
|
);
|
|
await this.#stores.sessionStore.discardStableConversationCopy(
|
|
header.id,
|
|
copy.requestFingerprint,
|
|
);
|
|
}
|
|
|
|
async #hasAdmittedRevisionTurn(sessionId: string): Promise<boolean> {
|
|
if (
|
|
(await this.#stores.agentRunStore.listRootTurnAdmissionsForRecovery(sessionId)).length > 0
|
|
) {
|
|
return true;
|
|
}
|
|
const messages = await this.#stores.sessionStore.readMessagesForRecovery(sessionId);
|
|
let boundary = -1;
|
|
for (let index = 0; index < messages.length; index += 1) {
|
|
const message = messages[index]!;
|
|
if (
|
|
message.type === 'system_note' &&
|
|
message.kind === 'session_start' &&
|
|
isRevisionStartData(message.data)
|
|
) {
|
|
boundary = index;
|
|
}
|
|
}
|
|
return boundary >= 0 && messages.slice(boundary + 1).some((message) => message.type === 'user');
|
|
}
|
|
|
|
async #hasCommittedConversationCopyDependent(sessionId: string): Promise<boolean> {
|
|
return (await this.#stores.sessionStore.listHeaders()).some(
|
|
(header) =>
|
|
header.conversationCopy?.state === 'committed' &&
|
|
header.conversationCopy.sourceSessionId === sessionId,
|
|
);
|
|
}
|
|
}
|
|
|
|
function isConversationRuntimeFactRewriteUnsupported(error: unknown): boolean {
|
|
return (
|
|
error instanceof Error &&
|
|
'code' in error &&
|
|
error.code === 'branch_runtime_fact_rewrite_unsupported'
|
|
);
|
|
}
|
|
|
|
function conversationCopyFingerprint(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
): `sha256:${string}` {
|
|
// The optimistic source revision guards only the initial create. Once this
|
|
// target exists, its stable identity must resolve the committed outcome even
|
|
// if a reconnecting Client observes a newer source revision.
|
|
const identity = [
|
|
kind === 'side_conversation' ? 'session.conversation-copy.v2' : 'session.conversation-copy.v1',
|
|
kind,
|
|
input.sourceSessionId,
|
|
input.targetSessionId,
|
|
input.sourceTurnId,
|
|
];
|
|
return `sha256:${createHash('sha256').update(JSON.stringify(identity)).digest('hex')}`;
|
|
}
|
|
|
|
function conversationCopyStartNote(
|
|
kind: ConversationCopySemanticKind,
|
|
input: SessionConversationCopyInput,
|
|
createInput: ConversationCopyCreateInput,
|
|
): StoredMessage {
|
|
return {
|
|
type: 'system_note',
|
|
id: randomUUID(),
|
|
ts: Date.now(),
|
|
kind: 'session_start',
|
|
data:
|
|
kind !== 'revision'
|
|
? {
|
|
parentSessionId: input.sourceSessionId,
|
|
branchOfTurnId: input.sourceTurnId,
|
|
}
|
|
: {
|
|
revisionRootSessionId: createInput.revisionRootSessionId,
|
|
revisionParentSessionId: input.sourceSessionId,
|
|
revisionOfTurnId: input.sourceTurnId,
|
|
revisionIndex: createInput.revisionIndex,
|
|
revisionState: 'preparing',
|
|
},
|
|
};
|
|
}
|
|
|
|
function conversationCopySemanticKind(
|
|
kind: ConversationCopyKind,
|
|
input: SessionConversationCopyInput,
|
|
): ConversationCopySemanticKind {
|
|
return kind === 'branch' && input.intent === 'side_conversation' ? input.intent : kind;
|
|
}
|
|
|
|
function persistedConversationCopyKind(kind: ConversationCopySemanticKind): ConversationCopyKind {
|
|
return kind === 'revision' ? 'revision' : 'branch';
|
|
}
|
|
|
|
function isRevisionStartData(value: unknown): boolean {
|
|
return (
|
|
!!value &&
|
|
typeof value === 'object' &&
|
|
!Array.isArray(value) &&
|
|
'revisionRootSessionId' in value
|
|
);
|
|
}
|
|
|
|
function collectArchivedToolResultPlaceholders(
|
|
events: readonly RuntimeEvent[],
|
|
messages: readonly StoredMessage[],
|
|
copyTurnIds: readonly string[],
|
|
operationalEvents: readonly AgentRunEvent[],
|
|
): ArchivedToolResultCopyDescriptor[] | null {
|
|
const retainedTurnIds = new Set(copyTurnIds);
|
|
const archives = new Map<string, ArchivedToolResultCopyDescriptor>();
|
|
const add = (value: unknown): boolean => {
|
|
if (!isRecord(value) || value.kind !== 'maka.archived_tool_result') return true;
|
|
if (!isArchivedToolResultPlaceholder(value)) return false;
|
|
addDescriptor(value);
|
|
return true;
|
|
};
|
|
const addDescriptor = (descriptor: ArchivedToolResultCopyDescriptor): void => {
|
|
const key = `${descriptor.artifactId}:${descriptor.bodySha256}:${descriptor.originalBytes}`;
|
|
if (!archives.has(key)) archives.set(key, descriptor);
|
|
};
|
|
|
|
for (const event of events) {
|
|
if (retainedTurnIds.has(event.turnId) && event.content?.kind === 'function_response') {
|
|
if (!add(event.content.result)) return null;
|
|
}
|
|
}
|
|
// A pruned result's body is now named by its durable transition rather than
|
|
// by the RuntimeEvent, so the copy must reach the ledger to find it. Missing
|
|
// this is not a cosmetic gap: the target Session would carry a placeholder
|
|
// pointing at an artifact that was never copied.
|
|
for (const event of operationalEvents) {
|
|
if (event.type !== MODEL_PROJECTION_TRANSITION_EVENT_TYPE) continue;
|
|
let transition: ModelProjectionTransition;
|
|
try {
|
|
transition = decodeModelProjectionTransition(event.data?.transition, event.sessionId);
|
|
} catch {
|
|
return null;
|
|
}
|
|
if (transition.replacement.kind !== 'json') return null;
|
|
if (!add(transition.replacement.value)) return null;
|
|
}
|
|
for (const message of messages) {
|
|
if (message.type !== 'tool_result') continue;
|
|
if (message.content.kind === 'json') {
|
|
if (!add(message.content.value)) return null;
|
|
continue;
|
|
}
|
|
if (message.content.kind === 'archived_tool_result') {
|
|
if (!message.content.artifactId && !message.content.bodySha256) continue;
|
|
if (!message.content.artifactId || !message.content.bodySha256) return null;
|
|
addDescriptor({
|
|
artifactId: message.content.artifactId,
|
|
bodySha256: message.content.bodySha256,
|
|
originalBytes: message.content.originalBytes,
|
|
});
|
|
}
|
|
}
|
|
return [...archives.values()];
|
|
}
|
|
|
|
interface ArchivedToolResultCopyDescriptor {
|
|
readonly artifactId: string;
|
|
readonly bodySha256: string;
|
|
readonly originalBytes: number;
|
|
}
|
|
|
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
return !!value && typeof value === 'object' && !Array.isArray(value);
|
|
}
|
|
|
|
function copySuccess(result: SessionConversationCopyResult): ConversationCopyOutcome {
|
|
return { ok: true, result };
|
|
}
|
|
|
|
function abandonSuccess(result: SessionRevisionAbandonResult): RevisionAbandonOutcome {
|
|
return { ok: true, result };
|
|
}
|
|
|
|
function copyFailure(
|
|
code: Extract<ConversationCopyOutcome, { ok: false }>['error']['code'],
|
|
message: string,
|
|
): ConversationCopyOutcome {
|
|
return { ok: false, error: { code, message } };
|
|
}
|
|
|
|
function abandonFailure(
|
|
code: Extract<RevisionAbandonOutcome, { ok: false }>['error']['code'],
|
|
message: string,
|
|
): RevisionAbandonOutcome {
|
|
return { ok: false, error: { code, message } };
|
|
}
|