Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 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 @@ -420,6 +420,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
Loading