packages/runtime-host/src/server/execution-artifacts.ts - maka - Git at Google

 /*
  * 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 } from 'node:crypto';
 import { open, realpath, stat } from 'node:fs/promises';
 import { isAbsolute, resolve } from 'node:path';
 import { MAX_ATTACHMENT_BYTES } from '@maka/core/attachments';
 import {
   createToolResultArchiveCapability,
   type ToolResultArchiveCapability,
   type ToolResultArchiveRecorder,
 } from '@maka/runtime/tool-result-archive-capability';
 import { isPathInside } from '@maka/runtime/path-containment';
 import {
   createLedgerArchivePreparer,
   createLedgerArchiveResourceReader,
 } from '@maka/runtime/ledger-tool-result-archive-reader';
 import type { ToolResultArchiveEvidenceReader } from '@maka/core/tool-result-archive-evidence';
 import { type ToolArtifactRecorderInput } from '@maka/runtime/tool-artifacts';
 import {
   type ToolResultArchiveReaderInput,
   type ToolResultArchiveReadResult,
 } from '@maka/runtime/context-budget';
 import { type ToolResultArchiveResourceReadInput } from '@maka/runtime/tool-result-archive-resource';
 import type { InteractiveArtifactStoreWriter } from '@maka/storage/artifact-stores';
 import type { SessionManagerDeps } from '@maka/runtime/session-manager';
 import type { SessionAdmissionGate } from './session-admission-gate.js';
 import type { SessionPresenceReader } from './session-presence.js';

 export interface HostExecutionArtifactServices {
   recordToolArtifacts(event: ToolArtifactRecorderInput): Promise<void>;
   publishChildWorkspacePatch: NonNullable<SessionManagerDeps['publishChildWorkspacePatch']>;
   /**
    * New archives use the Session ledger. Legacy Artifact refs retain their
    * scoped reader; the writer never falls back to publishing Artifact bytes.
    */
   toolResultArchive: ToolResultArchiveCapability;
 }

 export function createHostExecutionArtifactServices(input: {
   artifacts: InteractiveArtifactStoreWriter;
   requestDrain: () => void;
   sessionAdmission: SessionAdmissionGate;
   sessions: SessionPresenceReader;
   archiveEvidence?: ToolResultArchiveEvidenceReader;
 }): HostExecutionArtifactServices {
   const runWrite = async <T>(operation: () => Promise<T>): Promise<T> => {
     try {
       return await operation();
     } catch (error) {
       input.requestDrain();
       throw error;
     }
   };

   const publish = (artifact: Parameters<InteractiveArtifactStoreWriter['create']>[0]) =>
     input.sessionAdmission.runOrJoin(artifact.sessionId, async () => {
       if ((await input.sessions.probeSessionRemoval(artifact.sessionId)).kind !== 'present')
         return null;
       return runWrite(() => input.artifacts.create(artifact));
     });

   const recordToolArtifacts = async (event: ToolArtifactRecorderInput): Promise<void> => {
     for (const candidate of event.candidates) {
       let content = candidate.content;
       if (content === undefined && candidate.sourcePath) {
         content = (await readBoundedSourceFile(event.cwd, candidate.sourcePath)) ?? undefined;
       }
       if (content === undefined || contentBytes(content) > MAX_ATTACHMENT_BYTES) continue;
       await publish({
         sessionId: event.sessionId,
         turnId: event.turnId,
         name: candidate.name,
         kind: candidate.kind,
         content,
         ...(candidate.mimeType ? { mimeType: candidate.mimeType } : {}),
         source: candidate.source ?? 'tool_result',
         ...(candidate.summary ? { summary: candidate.summary } : {}),
       });
     }
   };

   const prepareLedger = input.archiveEvidence
     ? createLedgerArchivePreparer(input.archiveEvidence)
     : undefined;
   const readLedgerResource = input.archiveEvidence
     ? createLedgerArchiveResourceReader(input.archiveEvidence)
     : undefined;
   const prepareLedgerForCommit: ToolResultArchiveRecorder = async (event) => {
     const accepted = { ...event };
     if (!prepareLedger || !(await prepareLedger(accepted))) return;
     return {
       ledger: true,
       commitTransition: (transition, persist) =>
         input.sessionAdmission.runOrJoin(accepted.sessionId, async () => {
           if (
             (await input.sessions.probeSessionRemoval(accepted.sessionId)).kind !== 'present' ||
             !(await prepareLedger(accepted))
           )
             return false;
           await persist(transition);
           return true;
         }),
     };
   };
   const services: HostExecutionArtifactServices = {
     recordToolArtifacts,
     publishChildWorkspacePatch: async ({ sessionId, turnId, binding, patch }) => {
       const artifact = await publish({
         id: subagentWritebackArtifactId(sessionId, turnId),
         sessionId,
         turnId,
         name: 'workspace.patch',
         kind: 'diff',
         content: patch,
         mimeType: 'text/x-diff; charset=utf-8',
         source: 'subagent_writeback',
         summary: `Workspace changes relative to ${binding.baseCommit}.`,
       });
       if (!artifact)
         throw new Error(`Child Session ${sessionId} was retired before patch publication`);
       return artifact;
     },
     toolResultArchive: createToolResultArchiveCapability({
       archiveToolResult: prepareLedgerForCommit,
       readArchivedToolResultResource: (event: ToolResultArchiveResourceReadInput) =>
         event.storage === 'ledger' || event.storage === 'event'
           ? (readLedgerResource?.(event) ?? { ok: false, reason: 'read_failed' })
           : readArchive(input.artifacts, event),
     }),
   };
   return Object.freeze(services);
 }

 function subagentWritebackArtifactId(sessionId: string, turnId: string): string {
   const digest = createHash('sha256')
     .update('maka-subagent-writeback-v1\0')
     .update(sessionId)
     .update('\0')
     .update(turnId)
     .digest('hex')
     .slice(0, 32);
   return `subagent_writeback_${digest}`;
 }

 async function readBoundedSourceFile(cwd: string, sourcePath: string): Promise<Buffer | null> {
   const candidate = isAbsolute(sourcePath) ? sourcePath : resolve(cwd, sourcePath);
   let root: string;
   let target: string;
   try {
     [root, target] = await Promise.all([realpath(cwd), realpath(candidate)]);
   } catch {
     return null;
   }
   if (!isPathInside(root, target)) return null;
   let handle: Awaited<ReturnType<typeof open>> | undefined;
   try {
     handle = await open(target, 'r');
     const opened = await handle.stat();
     if (!opened.isFile() || opened.size > MAX_ATTACHMENT_BYTES) return null;
     const currentTarget = await realpath(target);
     if (!isPathInside(root, currentTarget)) return null;
     const current = await stat(currentTarget);
     if (current.dev !== opened.dev || current.ino !== opened.ino) return null;
     const bytes = Buffer.allocUnsafe(opened.size);
     let offset = 0;
     while (offset < bytes.byteLength) {
       const read = await handle.read(bytes, offset, bytes.byteLength - offset, offset);
       if (read.bytesRead === 0) break;
       offset += read.bytesRead;
     }
     return offset === bytes.byteLength ? bytes : bytes.subarray(0, offset);
   } catch {
     return null;
   } finally {
     await handle?.close().catch(() => undefined);
   }
 }

 function contentBytes(content: string | Uint8Array): number {
   return typeof content === 'string' ? Buffer.byteLength(content, 'utf8') : content.byteLength;
 }

 async function readArchive(
   artifacts: InteractiveArtifactStoreWriter,
   event: Pick<
     Extract<ToolResultArchiveReaderInput, { rewriteVersion: 1 }>,
     'artifactId' | 'sessionId' | 'bodySha256' | 'originalBytes' | 'maxBytes'
   >,
 ): Promise<ToolResultArchiveReadResult> {
   const entry = await artifacts.getInSession(event.sessionId, event.artifactId);
   const record = entry.record;
   if (!record) return { ok: false, reason: 'not_found' };
   if (record.source !== 'tool_result_archive') return { ok: false, reason: 'source_mismatch' };
   if (record.sizeBytes !== event.originalBytes) return { ok: false, reason: 'size_mismatch' };
   const read = await artifacts.readTextInSession(event.sessionId, event.artifactId, {
     maxBytes: event.maxBytes ?? event.originalBytes,
   });
   if (!read.ok) return read;
   if (sha256(read.text) !== event.bodySha256) return { ok: false, reason: 'corrupt' };
   return { ok: true, serializedResult: read.text };
 }

 function sha256(text: string): string {
   return createHash('sha256').update(text).digest('hex');
 }
	/*
	* 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 } from 'node:crypto';
	import { open, realpath, stat } from 'node:fs/promises';
	import { isAbsolute, resolve } from 'node:path';
	import { MAX_ATTACHMENT_BYTES } from '@maka/core/attachments';
	import {
	createToolResultArchiveCapability,
	type ToolResultArchiveCapability,
	type ToolResultArchiveRecorder,
	} from '@maka/runtime/tool-result-archive-capability';
	import { isPathInside } from '@maka/runtime/path-containment';
	import {
	createLedgerArchivePreparer,
	createLedgerArchiveResourceReader,
	} from '@maka/runtime/ledger-tool-result-archive-reader';
	import type { ToolResultArchiveEvidenceReader } from '@maka/core/tool-result-archive-evidence';
	import { type ToolArtifactRecorderInput } from '@maka/runtime/tool-artifacts';
	import {
	type ToolResultArchiveReaderInput,
	type ToolResultArchiveReadResult,
	} from '@maka/runtime/context-budget';
	import { type ToolResultArchiveResourceReadInput } from '@maka/runtime/tool-result-archive-resource';
	import type { InteractiveArtifactStoreWriter } from '@maka/storage/artifact-stores';
	import type { SessionManagerDeps } from '@maka/runtime/session-manager';
	import type { SessionAdmissionGate } from './session-admission-gate.js';
	import type { SessionPresenceReader } from './session-presence.js';

	export interface HostExecutionArtifactServices {
	recordToolArtifacts(event: ToolArtifactRecorderInput): Promise<void>;
	publishChildWorkspacePatch: NonNullable<SessionManagerDeps['publishChildWorkspacePatch']>;
	/**
	* New archives use the Session ledger. Legacy Artifact refs retain their
	* scoped reader; the writer never falls back to publishing Artifact bytes.
	*/
	toolResultArchive: ToolResultArchiveCapability;
	}

	export function createHostExecutionArtifactServices(input: {
	artifacts: InteractiveArtifactStoreWriter;
	requestDrain: () => void;
	sessionAdmission: SessionAdmissionGate;
	sessions: SessionPresenceReader;
	archiveEvidence?: ToolResultArchiveEvidenceReader;
	}): HostExecutionArtifactServices {
	const runWrite = async <T>(operation: () => Promise<T>): Promise<T> => {
	try {
	return await operation();
	} catch (error) {
	input.requestDrain();
	throw error;
	}
	};

	const publish = (artifact: Parameters<InteractiveArtifactStoreWriter['create']>[0]) =>
	input.sessionAdmission.runOrJoin(artifact.sessionId, async () => {
	if ((await input.sessions.probeSessionRemoval(artifact.sessionId)).kind !== 'present')
	return null;
	return runWrite(() => input.artifacts.create(artifact));
	});

	const recordToolArtifacts = async (event: ToolArtifactRecorderInput): Promise<void> => {
	for (const candidate of event.candidates) {
	let content = candidate.content;
	if (content === undefined && candidate.sourcePath) {
	content = (await readBoundedSourceFile(event.cwd, candidate.sourcePath)) ?? undefined;
	}
	if (content === undefined \|\| contentBytes(content) > MAX_ATTACHMENT_BYTES) continue;
	await publish({
	sessionId: event.sessionId,
	turnId: event.turnId,
	name: candidate.name,
	kind: candidate.kind,
	content,
	...(candidate.mimeType ? { mimeType: candidate.mimeType } : {}),
	source: candidate.source ?? 'tool_result',
	...(candidate.summary ? { summary: candidate.summary } : {}),
	});
	}
	};

	const prepareLedger = input.archiveEvidence
	? createLedgerArchivePreparer(input.archiveEvidence)
	: undefined;
	const readLedgerResource = input.archiveEvidence
	? createLedgerArchiveResourceReader(input.archiveEvidence)
	: undefined;
	const prepareLedgerForCommit: ToolResultArchiveRecorder = async (event) => {
	const accepted = { ...event };
	if (!prepareLedger \|\| !(await prepareLedger(accepted))) return;
	return {
	ledger: true,
	commitTransition: (transition, persist) =>
	input.sessionAdmission.runOrJoin(accepted.sessionId, async () => {
	if (
	(await input.sessions.probeSessionRemoval(accepted.sessionId)).kind !== 'present' \|\|
	!(await prepareLedger(accepted))
	)
	return false;
	await persist(transition);
	return true;
	}),
	};
	};
	const services: HostExecutionArtifactServices = {
	recordToolArtifacts,
	publishChildWorkspacePatch: async ({ sessionId, turnId, binding, patch }) => {
	const artifact = await publish({
	id: subagentWritebackArtifactId(sessionId, turnId),
	sessionId,
	turnId,
	name: 'workspace.patch',
	kind: 'diff',
	content: patch,
	mimeType: 'text/x-diff; charset=utf-8',
	source: 'subagent_writeback',
	summary: `Workspace changes relative to ${binding.baseCommit}.`,
	});
	if (!artifact)
	throw new Error(`Child Session ${sessionId} was retired before patch publication`);
	return artifact;
	},
	toolResultArchive: createToolResultArchiveCapability({
	archiveToolResult: prepareLedgerForCommit,
	readArchivedToolResultResource: (event: ToolResultArchiveResourceReadInput) =>
	event.storage === 'ledger' \|\| event.storage === 'event'
	? (readLedgerResource?.(event) ?? { ok: false, reason: 'read_failed' })
	: readArchive(input.artifacts, event),
	}),
	};
	return Object.freeze(services);
	}

	function subagentWritebackArtifactId(sessionId: string, turnId: string): string {
	const digest = createHash('sha256')
	.update('maka-subagent-writeback-v1\0')
	.update(sessionId)
	.update('\0')
	.update(turnId)
	.digest('hex')
	.slice(0, 32);
	return `subagent_writeback_${digest}`;
	}

	async function readBoundedSourceFile(cwd: string, sourcePath: string): Promise<Buffer \| null> {
	const candidate = isAbsolute(sourcePath) ? sourcePath : resolve(cwd, sourcePath);
	let root: string;
	let target: string;
	try {
	[root, target] = await Promise.all([realpath(cwd), realpath(candidate)]);
	} catch {
	return null;
	}
	if (!isPathInside(root, target)) return null;
	let handle: Awaited<ReturnType<typeof open>> \| undefined;
	try {
	handle = await open(target, 'r');
	const opened = await handle.stat();
	if (!opened.isFile() \|\| opened.size > MAX_ATTACHMENT_BYTES) return null;
	const currentTarget = await realpath(target);
	if (!isPathInside(root, currentTarget)) return null;
	const current = await stat(currentTarget);
	if (current.dev !== opened.dev \|\| current.ino !== opened.ino) return null;
	const bytes = Buffer.allocUnsafe(opened.size);
	let offset = 0;
	while (offset < bytes.byteLength) {
	const read = await handle.read(bytes, offset, bytes.byteLength - offset, offset);
	if (read.bytesRead === 0) break;
	offset += read.bytesRead;
	}
	return offset === bytes.byteLength ? bytes : bytes.subarray(0, offset);
	} catch {
	return null;
	} finally {
	await handle?.close().catch(() => undefined);
	}
	}

	function contentBytes(content: string \| Uint8Array): number {
	return typeof content === 'string' ? Buffer.byteLength(content, 'utf8') : content.byteLength;
	}

	async function readArchive(
	artifacts: InteractiveArtifactStoreWriter,
	event: Pick<
	Extract<ToolResultArchiveReaderInput, { rewriteVersion: 1 }>,
	'artifactId' \| 'sessionId' \| 'bodySha256' \| 'originalBytes' \| 'maxBytes'
	>,
	): Promise<ToolResultArchiveReadResult> {
	const entry = await artifacts.getInSession(event.sessionId, event.artifactId);
	const record = entry.record;
	if (!record) return { ok: false, reason: 'not_found' };
	if (record.source !== 'tool_result_archive') return { ok: false, reason: 'source_mismatch' };
	if (record.sizeBytes !== event.originalBytes) return { ok: false, reason: 'size_mismatch' };
	const read = await artifacts.readTextInSession(event.sessionId, event.artifactId, {
	maxBytes: event.maxBytes ?? event.originalBytes,
	});
	if (!read.ok) return read;
	if (sha256(read.text) !== event.bodySha256) return { ok: false, reason: 'corrupt' };
	return { ok: true, serializedResult: read.text };
	}

	function sha256(text: string): string {
	return createHash('sha256').update(text).digest('hex');
	}