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
1 change: 1 addition & 0 deletions packages/core/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
"private": true,
"exports": {
"./durable-tool-result-projection": "./dist/durable-tool-result-projection.js",
"./model-projection-transition": "./dist/model-projection-transition.js",
"./canonical-runtime-event": "./dist/canonical-runtime-event.js",
"./runtime-boundary": "./dist/runtime-boundary.js",
"./runtime-event": "./dist/runtime-event.js",
Expand Down
107 changes: 107 additions & 0 deletions packages/core/src/__tests__/model-projection-transition.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
/*
* 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 assert from 'node:assert/strict';
import { describe, test } from 'node:test';

import type { DurableToolResultProjection } from '../durable-tool-result-projection.js';
import {
buildModelProjectionTransition,
decodeModelProjectionTransition,
durableToolResultProjectionDigest,
isModelProjectionTransition,
} from '../model-projection-transition.js';

const SOURCE: DurableToolResultProjection = {
version: 1,
kind: 'text',
text: 'a large tool result',
};

const REPLACEMENT: DurableToolResultProjection = {
version: 1,
kind: 'json',
value: { kind: 'maka.archived_tool_result', artifactId: 'artifact-1' },
};

function build(overrides: Partial<Parameters<typeof buildModelProjectionTransition>[0]> = {}) {
return buildModelProjectionTransition({
sessionId: 'session-1',
target: {
runtimeEventId: 'rt-result',
part: 'tool_result',
toolCallId: 'tool-1',
toolName: 'Read',
},
sourceProjection: SOURCE,
replacement: REPLACEMENT,
now: 1_700_000_000,
...overrides,
});
}

describe('model projection transition schema', () => {
test('digests the same projection identically regardless of key order', () => {
const reordered = {
kind: 'text',
text: SOURCE.text,
version: 1,
} as DurableToolResultProjection;
assert.equal(
durableToolResultProjectionDigest(reordered),
durableToolResultProjectionDigest(SOURCE),
);
});

test('binds the record to the projection it may replace', () => {
const transition = build();
assert.equal(transition.sourceProjectionDigest, durableToolResultProjectionDigest(SOURCE));
assert.equal(transition.createdAt, 1_700_000_000);
});

test('derives one id from content, so a duplicated concurrent append is idempotent', () => {
assert.equal(build().transitionId, build().transitionId);
// The clock is not part of the decision, so it must not be part of the id.
assert.equal(build().transitionId, build({ now: 1_800_000_000 }).transitionId);
assert.notEqual(
build().transitionId,
build({ previousTransitionId: 'mptransition-earlier' }).transitionId,
);
});

test('rejects a record that belongs to another Session', () => {
const transition = build();
assert.ok(isModelProjectionTransition(transition, 'session-1'));
assert.equal(isModelProjectionTransition(transition, 'session-2'), false);
assert.throws(() => decodeModelProjectionTransition(transition, 'session-2'));
});

test('rejects an unknown field and an unrepresentable replacement', () => {
const transition = build();
assert.throws(() =>
decodeModelProjectionTransition({ ...transition, extra: true }, 'session-1'),
);
assert.throws(() =>
decodeModelProjectionTransition(
{ ...transition, replacement: { version: 1, kind: 'text' } },
'session-1',
),
);
});
});
1 change: 1 addition & 0 deletions packages/core/src/agent-run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -426,6 +426,7 @@ export const AGENT_RUN_EVENT_TYPES = [
'provider_request_attempt_recorded',
'model_call_attempt_recorded',
'history_compact_checkpoint_recorded',
'model_projection_transition_recorded',
'task_gate_decided',
'abort_requested',
'run_completed',
Expand Down
8 changes: 7 additions & 1 deletion packages/core/src/events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -794,7 +794,13 @@ export type ToolResultContent =
originalEstimatedTokens: number;
originalBytes: number;
rewriteVersion: number;
reason: 'stale_tool_result_pruned_before_compact';
/**
* Both prune paths now record the same durable projection transition
* (#4283), so the archived-result read model spans both reasons.
*/
reason:
| 'stale_tool_result_pruned_before_compact'
| 'active_current_turn_tool_result_pruned_before_next_step';
}
| {
kind: 'terminal';
Expand Down
226 changes: 226 additions & 0 deletions packages/core/src/model-projection-transition.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,226 @@
/*
* 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.
*/

/**
* Durable model-projection transitions (#4283).
*
* A successful model-visible history is append-only. Any lossy change to
* already-visible history — pruning a large Tool Result, omitting an image a
* provider rejected — must first become a durable successor in the append-only
* operational AgentRunEvent ledger, so no later replay, compaction, branch, or
* restart can restore the replaced form.
*
* This module owns the one typed record that expresses such a change. It is
* sparse — it names one projection part of one RuntimeEvent — and so is not a
* generalization of the contiguous-prefix `HistoryCompactCheckpoint` (#4283).
*
* Everything a deterministic reduction needs is on the record:
*
* - `target` — which RuntimeEvent projection part is replaced;
* - `sourceProjectionDigest` — the exact projection it is allowed to replace,
* so a stale concurrent writer cannot apply against content it never saw;
* - `replacement` — what the model sees instead, including where the replaced
* body still lives when it is recoverable at all;
* - `previousTransitionId` — the predecessor for this target, which is also the
* reduction's ordering authority: readers follow the chain rather than a
* cursor, so ledger order and wall-clock skew cannot change the result.
*/

import * as nodeCrypto from 'node:crypto';

import {
decodeDurableToolResultProjection,
type DurableToolResultProjection,
} from './durable-tool-result-projection.js';
import { stableJsonStringify } from './tool-args-identity.js';
import { defineObjectShape, hasExactShape, isFiniteNumber, isRecord } from './record-schema.js';

export const MODEL_PROJECTION_TRANSITION_VERSION = 1 as const;

/** The append-only operational ledger record that carries one transition. */
export const MODEL_PROJECTION_TRANSITION_EVENT_TYPE = 'model_projection_transition_recorded';

/**
* The addressed projection part. `tool_result` is the whole durable Tool Result
* projection of one `function_response` RuntimeEvent — the only part kind that
* exists while the projection schema has no independently addressable segments.
*/
export interface ModelProjectionTransitionTarget {
runtimeEventId: string;
part: 'tool_result';
toolCallId: string;
toolName: string;
}

export interface ModelProjectionTransition {
kind: 'maka.model_projection_transition';
version: typeof MODEL_PROJECTION_TRANSITION_VERSION;
transitionId: string;
sessionId: string;
createdAt: number;
target: ModelProjectionTransitionTarget;
/** Digest of the projection this record is allowed to replace. */
sourceProjectionDigest: `sha256:${string}`;
replacement: DurableToolResultProjection;
/**
* The transition this one supersedes for the same target, if any.
*
* Absent means "applies to the base projection". Together with
* `sourceProjectionDigest` this is the only ordering a reducer needs.
*/
previousTransitionId?: string;
}

const TRANSITION_SHAPE = defineObjectShape<ModelProjectionTransition>()(
[
'kind',
'version',
'transitionId',
'sessionId',
'createdAt',
'target',
'sourceProjectionDigest',
'replacement',
],
['previousTransitionId'],
);

const TARGET_SHAPE = defineObjectShape<ModelProjectionTransitionTarget>()(
['runtimeEventId', 'part', 'toolCallId', 'toolName'],
[],
);

/**
* The identity of one durable projection, over strict key-sorted JSON.
*
* Writer and reducer must agree byte for byte: a digest computed one way at
* write time and another at read time would silently turn every transition
* into a source mismatch, i.e. into content that quietly comes back.
*/
export function durableToolResultProjectionDigest(
projection: DurableToolResultProjection,
): `sha256:${string}` {
return `sha256:${nodeCrypto
.createHash('sha256')
.update(stableJsonStringify(projection))
.digest('hex')}`;
}

export interface BuildModelProjectionTransitionInput {
sessionId: string;
target: ModelProjectionTransitionTarget;
sourceProjection: DurableToolResultProjection;
replacement: DurableToolResultProjection;
previousTransitionId?: string;
now: number;
}

/**
* Build one transition with a content-derived id.
*
* The id digests everything the record asserts and nothing about when or where
* it was written, so two writers that independently decide the same replacement
* for the same source produce the same record: a duplicate concurrent append is
* idempotent rather than a second competing successor.
*/
export function buildModelProjectionTransition(
input: BuildModelProjectionTransitionInput,
): ModelProjectionTransition {
const sourceProjectionDigest = durableToolResultProjectionDigest(
decodeDurableToolResultProjection(input.sourceProjection),
);
const replacement = decodeDurableToolResultProjection(input.replacement);
const body = {
version: MODEL_PROJECTION_TRANSITION_VERSION,
sessionId: input.sessionId,
target: input.target,
sourceProjectionDigest,
replacement,
...(input.previousTransitionId ? { previousTransitionId: input.previousTransitionId } : {}),
};
const transitionId = `mptransition-${nodeCrypto
.createHash('sha256')
.update(stableJsonStringify(body))
.digest('hex')
.slice(0, 32)}`;
return decodeModelProjectionTransition(
{
kind: 'maka.model_projection_transition',
transitionId,
createdAt: input.now,
...body,
},
input.sessionId,
);
}

export function decodeModelProjectionTransition(
value: unknown,
sessionId: string,
): ModelProjectionTransition {
if (!isModelProjectionTransition(value, sessionId)) {
throw new Error('Invalid model projection transition');
}
return value;
}

export function isModelProjectionTransition(
value: unknown,
sessionId: string,
): value is ModelProjectionTransition {
if (
!isRecord(value) ||
!hasExactShape(value, TRANSITION_SHAPE) ||
value.kind !== 'maka.model_projection_transition' ||
value.version !== MODEL_PROJECTION_TRANSITION_VERSION ||
!nonEmptyString(value.transitionId) ||
value.sessionId !== sessionId ||
!isFiniteNumber(value.createdAt) ||
!isSha256Digest(value.sourceProjectionDigest) ||
(value.previousTransitionId !== undefined && !nonEmptyString(value.previousTransitionId)) ||
!isTransitionTarget(value.target)
) {
return false;
}
try {
decodeDurableToolResultProjection(value.replacement);
} catch {
return false;
}
return true;
}

function isTransitionTarget(value: unknown): value is ModelProjectionTransitionTarget {
return (
isRecord(value) &&
hasExactShape(value, TARGET_SHAPE) &&
nonEmptyString(value.runtimeEventId) &&
value.part === 'tool_result' &&
nonEmptyString(value.toolCallId) &&
nonEmptyString(value.toolName)
);
}

function isSha256Digest(value: unknown): value is `sha256:${string}` {
return typeof value === 'string' && /^sha256:[a-f0-9]{64}$/.test(value);
}

function nonEmptyString(value: unknown): value is string {
return typeof value === 'string' && value.length > 0;
}
3 changes: 2 additions & 1 deletion packages/core/src/tool-result-record-schema.ts
Original file line number Diff line number Diff line change
Expand Up @@ -246,7 +246,8 @@ function isNonShellToolResultContent(value: unknown): value is ToolResultContent
isFiniteNumber(value.originalEstimatedTokens) &&
isFiniteNumber(value.originalBytes) &&
isFiniteNumber(value.rewriteVersion) &&
value.reason === 'stale_tool_result_pruned_before_compact'
(value.reason === 'stale_tool_result_pruned_before_compact' ||
value.reason === 'active_current_turn_tool_result_pruned_before_next_step')
);
case 'image':
return (
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ export class HostDeepResearchCoordinator {
readText: (artifactId, options) =>
this.#artifacts.readTextInSession(sessionId, artifactId, options),
delete: (artifactId) =>
this.#artifacts.deleteOwnedDeepResearchArtifactInSession(sessionId, artifactId),
this.#artifacts.deleteOwnedArtifactInSession(sessionId, artifactId, 'deep_research'),
},
});
}
Expand Down
Loading