You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by ku...@apache.org on 2026/08/29 10:43:29 UTC
(maka) 01/01: refactor(storage): add context offload root authority
This is an automated email from the ASF dual-hosted git repository.
likun666661 pushed a commit to branch refactor/context-offload-root-authority
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 1b04f66df200c9e453ec952b00f220e06c52d430
Author: likun666661 <90...@users.noreply.github.com>
AuthorDate: Sat Aug 29 18:43:20 2026 +0800
refactor(storage): add context offload root authority
Generated-by: OpenAI Codex
---
.../src/__tests__/context-offload-store.test.ts | 228 ++++++++++++++++++
packages/storage/src/context-offload-store.ts | 264 +++++++++++++++++++++
.../storage/src/sqlite-context-offload-store.ts | 1 +
3 files changed, 493 insertions(+)
diff --git a/packages/storage/src/__tests__/context-offload-store.test.ts b/packages/storage/src/__tests__/context-offload-store.test.ts
new file mode 100644
index 000000000..604ea94c6
--- /dev/null
+++ b/packages/storage/src/__tests__/context-offload-store.test.ts
@@ -0,0 +1,228 @@
+/*
+ * 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 { mkdtemp, rm, stat } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { after, test } from 'node:test';
+import type { ContextOffloadLimits } from '@maka/core/context-offload';
+import {
+ authenticateInteractiveContextOffloadWriter,
+ openInteractiveContextOffloadStoreForWrite,
+ type InteractiveContextOffloadWriter,
+} from '../context-offload-store.js';
+import {
+ resolveStorageRoot,
+ StorageRootAuthorityError,
+ tryAcquireInteractiveRootOwner,
+ type InteractiveRootOwner,
+ type StorageRootLease,
+} from '../root-authority.js';
+import { CONTEXT_OFFLOAD_DATABASE_NAME } from '../sqlite-context-offload-store.js';
+import {
+ removeTrackedControlDirectories,
+ trackControlDirectory,
+} from './fixtures/control-directory-hygiene.js';
+
+after(removeTrackedControlDirectories);
+
+test('requires authentic Storage Root leases and writer facades', async () => {
+ await assert.rejects(
+ () =>
+ openInteractiveContextOffloadStoreForWrite({} as StorageRootLease<'interactive', 'write'>, {
+ limits: testLimits(),
+ }),
+ invalidLease,
+ );
+ assert.throws(
+ () => authenticateInteractiveContextOffloadWriter({} as InteractiveContextOffloadWriter),
+ invalidLease,
+ );
+});
+
+test('single-flights one limit-bound writer and snapshots admitted inputs', async () => {
+ await withInteractiveOwner(async (owner, root) => {
+ const mutableLimits = testLimits();
+ const opening = openInteractiveContextOffloadStoreForWrite(owner.lease, {
+ limits: mutableLimits,
+ });
+ (mutableLimits.ownerMaxBytes as { tool_result_archive: number }).tool_result_archive = 0;
+ (mutableLimits as { sessionLogicalBytes: number }).sessionLogicalBytes = 0;
+ const first = await opening;
+ try {
+ const second = await openInteractiveContextOffloadStoreForWrite(owner.lease, {
+ limits: testLimits(),
+ });
+ assert.strictEqual(second, first);
+ assert.strictEqual(authenticateInteractiveContextOffloadWriter(first), first);
+ assert.equal((await stat(join(root, CONTEXT_OFFLOAD_DATABASE_NAME))).isFile(), true);
+ await assert.rejects(
+ () =>
+ openInteractiveContextOffloadStoreForWrite(owner.lease, {
+ limits: { ...testLimits(), workspacePhysicalBytes: 63 },
+ }),
+ /different limits/u,
+ );
+
+ const bytes = new TextEncoder().encode('safe');
+ const input = {
+ sessionId: 'source',
+ owner: { kind: 'tool_result_archive' as const, ownerId: 'source-owner' },
+ bytes,
+ mediaType: 'application/json',
+ };
+ const putting = first.put(input);
+ input.sessionId = 'mutated';
+ input.owner.ownerId = 'mutated-owner';
+ input.mediaType = 'text/plain';
+ bytes.fill(0x78);
+ const stored = await putting;
+ assert.equal(stored.ok, true);
+ if (!stored.ok) return;
+ assert.equal(stored.record.sessionId, 'source');
+ assert.equal(stored.record.owner.ownerId, 'source-owner');
+ assert.equal(stored.record.mediaType, 'application/json');
+ assert.deepEqual(
+ await first.read({ sessionId: 'source', refId: stored.record.refId, maxBytes: 64 }),
+ { ok: true, record: stored.record, bytes: new TextEncoder().encode('safe') },
+ );
+
+ const copyInput = {
+ sourceSessionId: 'source',
+ targetSessionId: 'target',
+ references: [
+ {
+ sourceRefId: stored.record.refId,
+ targetOwner: { kind: 'tool_result_archive' as const, ownerId: 'target-owner' },
+ },
+ ],
+ };
+ const copying = first.copyReferences(copyInput);
+ copyInput.targetSessionId = 'mutated-target';
+ copyInput.references[0]!.targetOwner.ownerId = 'mutated-target-owner';
+ const copied = await copying;
+ assert.equal(copied.ok, true);
+ if (!copied.ok) return;
+ assert.deepEqual(
+ await first.copyReferences({
+ sourceSessionId: 'source',
+ targetSessionId: 'target',
+ references: [
+ {
+ sourceRefId: stored.record.refId,
+ targetOwner: { kind: 'tool_result_archive', ownerId: 'target-owner' },
+ },
+ ],
+ }),
+ copied,
+ );
+ } finally {
+ await first.close();
+ }
+ });
+});
+
+test('close drains admitted work, revokes the facade, and permits a clean reopen', async () => {
+ await withInteractiveOwner(async (owner) => {
+ const writer = await openInteractiveContextOffloadStoreForWrite(owner.lease, {
+ limits: testLimits(),
+ });
+ const admitted = writer.put({
+ sessionId: 'session-1',
+ owner: { kind: 'read_image_snapshot', ownerId: 'read-1' },
+ bytes: new TextEncoder().encode('image'),
+ mediaType: 'image/png',
+ });
+ const closing = writer.close();
+ const reopening = openInteractiveContextOffloadStoreForWrite(owner.lease, {
+ limits: testLimits(),
+ });
+ assert.equal((await admitted).ok, true);
+ await closing;
+ await assert.rejects(writer.usage(), invalidLease);
+ assert.throws(() => authenticateInteractiveContextOffloadWriter(writer), invalidLease);
+
+ const reopened = await reopening;
+ try {
+ assert.notStrictEqual(reopened, writer);
+ assert.deepEqual(await reopened.usage('session-1'), {
+ references: 1,
+ logicalBytes: 5,
+ physicalBytes: 5,
+ });
+ } finally {
+ await reopened.close();
+ }
+ });
+});
+
+test('root-owner close revokes new context-offload operations', async () => {
+ const root = await mkdtemp(join(tmpdir(), 'maka-context-offload-authority-revoke-'));
+ const capability = trackControlDirectory(
+ await resolveStorageRoot({ path: root, kind: 'interactive' }),
+ );
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ if (!owner) return;
+ const writer = await openInteractiveContextOffloadStoreForWrite(owner.lease, {
+ limits: testLimits(),
+ });
+ try {
+ await owner.close();
+ await assert.rejects(writer.usage(), invalidLease);
+ } finally {
+ await writer.close();
+ await owner.close();
+ await rm(root, { recursive: true, force: true });
+ }
+});
+
+function testLimits(): ContextOffloadLimits {
+ return {
+ ownerMaxBytes: {
+ read_image_snapshot: 64,
+ tool_result_archive: 64,
+ },
+ sessionLogicalBytes: 64,
+ workspacePhysicalBytes: 64,
+ };
+}
+
+function invalidLease(error: unknown): boolean {
+ return error instanceof StorageRootAuthorityError && error.code === 'invalid_lease';
+}
+
+async function withInteractiveOwner(
+ run: (owner: InteractiveRootOwner, root: string) => Promise<void>,
+): Promise<void> {
+ const root = await mkdtemp(join(tmpdir(), 'maka-context-offload-authority-'));
+ const capability = trackControlDirectory(
+ await resolveStorageRoot({ path: root, kind: 'interactive' }),
+ );
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ if (!owner) return;
+ try {
+ await run(owner, root);
+ } finally {
+ await owner.close();
+ await rm(root, { recursive: true, force: true });
+ }
+}
diff --git a/packages/storage/src/context-offload-store.ts b/packages/storage/src/context-offload-store.ts
new file mode 100644
index 000000000..c3e4f07b2
--- /dev/null
+++ b/packages/storage/src/context-offload-store.ts
@@ -0,0 +1,264 @@
+/*
+ * 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 { join } from 'node:path';
+import type {
+ ContextOffloadLimits,
+ ContextOffloadOwner,
+ ContextOffloadStore,
+} from '@maka/core/context-offload';
+import {
+ assertStorageRootLease,
+ runWithStorageRootLease,
+ StorageRootAuthorityError,
+ type StorageRootLease,
+} from './root-authority.js';
+import {
+ CONTEXT_OFFLOAD_DATABASE_NAME,
+ SqliteContextOffloadStore,
+} from './sqlite-context-offload-store.js';
+
+const writerBrand: unique symbol = Symbol('InteractiveContextOffloadWriter');
+const writers = new WeakSet<object>();
+const writerByLease = new WeakMap<
+ object,
+ { readonly writer: InteractiveContextOffloadWriter; readonly limitsKey: string }
+>();
+const writerOpeningByLease = new WeakMap<
+ object,
+ { readonly pending: Promise<InteractiveContextOffloadWriter>; readonly limitsKey: string }
+>();
+const writerClosingByLease = new WeakMap<
+ object,
+ { readonly pending: Promise<void>; readonly limitsKey: string }
+>();
+
+export interface OpenInteractiveContextOffloadStoreOptions {
+ readonly limits: ContextOffloadLimits;
+}
+
+export interface InteractiveContextOffloadWriter extends Omit<ContextOffloadStore, 'close'> {
+ readonly kind: 'interactive';
+ readonly access: 'write';
+ readonly [writerBrand]: true;
+ close(): Promise<void>;
+}
+
+export function authenticateInteractiveContextOffloadWriter(
+ writer: InteractiveContextOffloadWriter,
+): InteractiveContextOffloadWriter {
+ if (!writers.has(writer)) {
+ throw new StorageRootAuthorityError(
+ 'invalid_lease',
+ 'Expected an authentic interactive context-offload writer',
+ );
+ }
+ return writer;
+}
+
+/**
+ * Opens context-offload storage through an authenticated interactive write
+ * lease. Production callers must use this facade instead of constructing the
+ * low-level SQLite Store directly.
+ */
+export async function openInteractiveContextOffloadStoreForWrite(
+ lease: StorageRootLease<'interactive', 'write'>,
+ options: OpenInteractiveContextOffloadStoreOptions,
+): Promise<InteractiveContextOffloadWriter> {
+ const limits = snapshotLimits(options.limits);
+ const limitsKey = serializeLimits(limits);
+ await assertStorageRootLease(lease, 'interactive', 'write');
+ const existing = writerByLease.get(lease);
+ if (existing) {
+ assertSameLimits(existing.limitsKey, limitsKey);
+ return existing.writer;
+ }
+ const opening = writerOpeningByLease.get(lease);
+ if (opening) {
+ assertSameLimits(opening.limitsKey, limitsKey);
+ return opening.pending;
+ }
+ const closing = writerClosingByLease.get(lease);
+ if (closing) {
+ assertSameLimits(closing.limitsKey, limitsKey);
+ await closing.pending;
+ return openInteractiveContextOffloadStoreForWrite(lease, { limits });
+ }
+
+ const pending = Promise.resolve().then(async () => {
+ let store: SqliteContextOffloadStore | undefined;
+ try {
+ store = await runWithStorageRootLease(
+ lease,
+ 'interactive',
+ 'write',
+ async (root) =>
+ new SqliteContextOffloadStore(join(root, CONTEXT_OFFLOAD_DATABASE_NAME), { limits }),
+ );
+ await assertStorageRootLease(lease, 'interactive', 'write');
+ const raced = writerByLease.get(lease);
+ if (raced) {
+ store.close();
+ assertSameLimits(raced.limitsKey, limitsKey);
+ return raced.writer;
+ }
+ const writer = createWriterFacade(lease, store, limitsKey);
+ writers.add(writer);
+ writerByLease.set(lease, { writer, limitsKey });
+ return writer;
+ } catch (error) {
+ store?.close();
+ throw error;
+ }
+ });
+ writerOpeningByLease.set(lease, { pending, limitsKey });
+ try {
+ return await pending;
+ } finally {
+ if (writerOpeningByLease.get(lease)?.pending === pending) {
+ writerOpeningByLease.delete(lease);
+ }
+ }
+}
+
+function createWriterFacade(
+ lease: StorageRootLease<'interactive', 'write'>,
+ store: SqliteContextOffloadStore,
+ limitsKey: string,
+): InteractiveContextOffloadWriter {
+ let closed = false;
+ let closeTask: Promise<void> | undefined;
+ const activeOperations = new Set<Promise<unknown>>();
+ const run = <T>(operation: () => Promise<T>): Promise<T> => {
+ if (closed) {
+ return Promise.reject(
+ new StorageRootAuthorityError('invalid_lease', 'Context-offload writer is closed'),
+ );
+ }
+ const pending = runWithStorageRootLease(lease, 'interactive', 'write', operation);
+ activeOperations.add(pending);
+ void pending.finally(() => activeOperations.delete(pending)).catch(() => undefined);
+ return pending;
+ };
+ const writer: InteractiveContextOffloadWriter = {
+ kind: 'interactive',
+ access: 'write',
+ [writerBrand]: true,
+ put: (input) => {
+ const accepted = Object.freeze({
+ sessionId: input.sessionId,
+ owner: Object.freeze({ ...input.owner }),
+ bytes: new Uint8Array(input.bytes),
+ mediaType: input.mediaType,
+ ...(input.expectedSha256 === undefined ? {} : { expectedSha256: input.expectedSha256 }),
+ });
+ return run(() => store.put(accepted));
+ },
+ read: (input) => {
+ const accepted = Object.freeze({ ...input });
+ return run(() => store.read(accepted));
+ },
+ copyReferences: (input) => {
+ const accepted = Object.freeze({
+ sourceSessionId: input.sourceSessionId,
+ targetSessionId: input.targetSessionId,
+ references: Object.freeze(
+ input.references.map((reference) =>
+ Object.freeze({
+ sourceRefId: reference.sourceRefId,
+ targetOwner: Object.freeze({ ...reference.targetOwner }),
+ }),
+ ),
+ ),
+ });
+ return run(() => store.copyReferences(accepted));
+ },
+ releaseReference: (input) => {
+ const accepted = Object.freeze({ ...input });
+ return run(() => store.releaseReference(accepted));
+ },
+ retireSession: (sessionId) => run(() => store.retireSession(sessionId)),
+ collectGarbage: (input) => {
+ const accepted = Object.freeze({ ...input });
+ return run(() => store.collectGarbage(accepted));
+ },
+ usage: (sessionId) => run(() => store.usage(sessionId)),
+ close: () => {
+ if (closeTask) return closeTask;
+ closed = true;
+ if (writerByLease.get(lease)?.writer === writer) writerByLease.delete(lease);
+ writers.delete(writer);
+ let pending!: Promise<void>;
+ pending = (async () => {
+ try {
+ await Promise.allSettled([...activeOperations]);
+ store.close();
+ } finally {
+ if (writerClosingByLease.get(lease)?.pending === pending) {
+ writerClosingByLease.delete(lease);
+ }
+ }
+ })();
+ closeTask = pending;
+ writerClosingByLease.set(lease, { pending, limitsKey });
+ return pending;
+ },
+ };
+ return Object.freeze(writer);
+}
+
+function snapshotLimits(limits: ContextOffloadLimits): ContextOffloadLimits {
+ const ownerMaxBytes = Object.freeze({
+ read_image_snapshot: readLimit(
+ limits.ownerMaxBytes?.read_image_snapshot,
+ 'Read image snapshot byte limit',
+ ),
+ tool_result_archive: readLimit(
+ limits.ownerMaxBytes?.tool_result_archive,
+ 'Tool Result archive byte limit',
+ ),
+ }) satisfies Readonly<Record<ContextOffloadOwner['kind'], number>>;
+ return Object.freeze({
+ ownerMaxBytes,
+ sessionLogicalBytes: readLimit(limits.sessionLogicalBytes, 'Session context quota'),
+ workspacePhysicalBytes: readLimit(limits.workspacePhysicalBytes, 'Workspace context quota'),
+ });
+}
+
+function readLimit(value: unknown, label: string): number {
+ if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) {
+ throw new Error(`${label} must be a non-negative safe integer`);
+ }
+ return value;
+}
+
+function serializeLimits(limits: ContextOffloadLimits): string {
+ return [
+ limits.ownerMaxBytes.read_image_snapshot,
+ limits.ownerMaxBytes.tool_result_archive,
+ limits.sessionLogicalBytes,
+ limits.workspacePhysicalBytes,
+ ].join(':');
+}
+
+function assertSameLimits(existing: string, requested: string): void {
+ if (existing !== requested) {
+ throw new Error('Context-offload writer is already bound to different limits for this lease');
+ }
+}
diff --git a/packages/storage/src/sqlite-context-offload-store.ts b/packages/storage/src/sqlite-context-offload-store.ts
index d886b1e21..e4f6be2d1 100644
--- a/packages/storage/src/sqlite-context-offload-store.ts
+++ b/packages/storage/src/sqlite-context-offload-store.ts
@@ -90,6 +90,7 @@ interface GarbageCandidateRow {
size_bytes: unknown;
}
+/** Low-level implementation; production callers must use the Storage Root authority facade. */
export class SqliteContextOffloadStore implements ContextOffloadStore {
readonly #database: DatabaseSync;
readonly #limits: ContextOffloadLimits;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org