fix(storage): version raw RocksDB primary writes

DESIGN.md

@@ -24,6 +24,12 @@ To match v4: `OpenDBIObject` sets `randomAccessStructure = isPrimary`, and `Reco
 - The struct **read** hook is left intact, so records a prior v5 already wrote in struct mode still decode (read-compat); only new writes switch to records mode.
 - Companion change in `structon`: `prepareStructures` saves the shared structures in the legacy plain-array form when there are no typed structs (instead of the `{named, typed}` Map), so the `Symbol.for('structures')` buffer for a records-mode `__dbis__` is also v4-decodable.
 
+## Raw RocksDB primary writes are versioned at the store boundary
+
+Classic msgpackr shared-record ids occupy `0x40–0x5f`; record id 2 is therefore `0x42`. RocksDB primary records also begin with `0x42` in normal operation because a current-era millisecond timestamp encoded as a big-endian float64 has that first byte. `RecordEncoder.decode()` cannot distinguish a prefix-less classic record whose structure id is `0x42` from a versioned record using only the persisted bytes.
+
+`PrimaryRocksDatabase.put()` and `putSync()` enforce the write-side invariant: if an initialized RocksDB encoder for a versioned primary write reaches the store without metadata already staged by `recordUpdater`, the wrapper stages a monotonic version and a zero-flags metadata word. The full 12-byte header is required; a timestamp alone would make the decoder consume the first four payload bytes as metadata. A positive, finite positional numeric version from legacy raw-put callers is retained verbatim. Metadata is cleared in `finally`, including binary writes that bypass `RecordEncoder`, so staged state cannot leak into the next store operation. The wrapper also resets the blob encoder's prior-write signal before a raw record encode so the row cannot inherit a stale `HAS_BLOBS` flag. Normal table writes arrive with metadata already staged and stay on their existing path without an additional allocation.
+
 ## Version gate at startup: downgrades prompt, and only the minor direction is confirmable
 
 `getVersionUpdateInfo()` (`dataLayer/hdbInfoController.ts`) compares the store's `data_version_num` (latest `system.hdb_info` record) against the binary's `packageJson.version` on every start. Data newer than binary by a **major** version → hard refusal. Newer by a **minor** version → `forceDowngradePrompt()` asks for confirmation; answering yes records the data version back down to the binary's version and boots (upgrade directives are deliberately additive/downgrade-compatible — see the struct-mode section above and `patchHdbSecretIsHashAttribute` in `upgrade/directives/5-2-0.ts`).

agent/session.ts

@@ -60,12 +60,7 @@ export async function createSession(opts: CreateSessionOpts): Promise<AgentSessi
 		createdAt: now,
 		updatedAt: now,
 	};
-	// Write through the Resource-level `put` (transactional), NOT `primaryStore.put`: the transactional
-	// path stages the version/local-timestamp metadata so the record is stored with the metadata prefix.
-	// A raw `primaryStore.put` writes a prefix-less record, which — when it begins with classic shared-
-	// structure record-id #2 (byte 0x42 == 66) — is misread by the RocksDB decode heuristic as a
-	// timestamp-prefixed record (8 bytes stripped → corrupt → "Could not find typed structure"). See
-	// RecordEncoder.ts:328-335.
+	// Use the transactional Resource path for table lifecycle semantics; raw puts are versioned at the store boundary (#1762).
 	await getAgentSessionTable().put(row);
 	return row;
 }

resources/PrimaryRocksDatabase.ts

@@ -3,7 +3,7 @@ import { RocksDatabase, type RocksDatabaseOptions, constants, type Store } from
 const FRESH_VERSION_FLAG = constants.FRESH_VERSION_FLAG;
 import { WeakLRUCache } from 'weak-lru-cache';
 import { when } from '../utility/when.ts';
-import { entryMap, METADATA, type Entry } from './RecordEncoder.ts';
+import { clearNextEncoding, entryMap, METADATA, stageRawPrimaryEncoding, type Entry } from './RecordEncoder.ts';
 
 /**
  * RocksDatabase subclass that owns all primary-store behaviour for Harper tables:
@@ -150,6 +150,16 @@ export class PrimaryRocksDatabase extends RocksDatabase {
 		return when(this.getEntry(id, { ...options, async: true }), (entry: Entry) => entry?.value);
 	}
 
+	put(id: any, value: any, options?: any): any {
+		this.#cache?.delete(id);
+		const staged = stageRawPrimaryEncoding(this.#enc, typeof options === 'number' ? options : options?.version);
+		try {
+			return super.put(id, value, options);
+		} finally {
+			if (staged) clearNextEncoding();
+		}
+	}
+
 	getRange(options?: any): any {
 		const iterable = super.getRange(options);
 		if (options?.valuesForKey) return iterable.map((v: any) => v?.value);
@@ -172,7 +182,12 @@ export class PrimaryRocksDatabase extends RocksDatabase {
 
 	putSync(id: any, value: any, options?: any): any {
 		this.#cache?.delete(id);
-		return super.putSync(id, value, options);
+		const staged = stageRawPrimaryEncoding(this.#enc, typeof options === 'number' ? options : options?.version);
+		try {
+			return super.putSync(id, value, options);
+		} finally {
+			if (staged) clearNextEncoding();
+		}
 	}
 
 	removeSync(id: any, options?: any): any {

resources/RecordEncoder.ts

@@ -27,6 +27,7 @@ import {
 	encodeBlobsWithFilePath,
 	findBlobsInObject,
 	getFileId,
+	resetBlobsWereEncoded,
 } from './blob.ts';
 import { getThisNodeId } from './nodeIdMapping.ts';
 import { recordAction } from './analytics/write.ts';
@@ -677,6 +678,22 @@ export function setNextEncoding(timestamp: number, metadata: number, expiresAt =
 	nodeIdAtNextEncoding = nodeId;
 	residencyIdAtNextEncoding = residencyId;
 }
+export function stageRawPrimaryEncoding(encoder: any, version?: number) {
+	if (
+		timestampNextEncoding !== 0 ||
+		metadataInNextEncoding >= 0 ||
+		encoder?.isRocksDB !== true ||
+		encoder.useVersions === false ||
+		encoder.autoVersion
+	)
+		return false;
+	setNextEncoding(
+		typeof version === 'number' && version > 0 && Number.isFinite(version) ? version : getNextMonotonicTime(),
+		0
+	);
+	resetBlobsWereEncoded();
+	return true;
+}
 /**
  * Reset the module-level "next encoding" metadata to its no-metadata defaults. These globals are
  * set just before a versioned encode and consumed (and reset) by the encode hook. If the consuming

resources/blob.ts

@@ -133,6 +133,9 @@ let encodeForStorageForRecordId: number = undefined; // only enable encoding of
 let promisedWrites: Array<Promise<void>>;
 let currentStore: any; // the root store of the database we are currently encoding for
 export let blobsWereEncoded = false; // keep track of whether blobs were encoded with file paths
+export function resetBlobsWereEncoded() {
+	blobsWereEncoded = false;
+}
 // the header is 8 bytes
 const DEFAULT_BLOB_READ_TIMEOUT = 20000;
 /**

unitTests/resources/primaryRocksMetadataRepair.test.js

@@ -10,11 +10,22 @@ const fs = require('fs-extra');
 const { setupTestDBPath } = require('../testUtils');
 const { RocksDatabase } = require('@harperfast/rocksdb-js');
 const { Packr } = require('msgpackr');
-const { RecordEncoder, RecordObject, setNextEncoding } = require('#src/resources/RecordEncoder');
+const { ACTION_32_BIT } = require('#src/resources/auditStore');
+const { createBlob, encodeBlobsWithFilePath } = require('#src/resources/blob');
+const {
+	RecordEncoder,
+	RecordObject,
+	setNextEncoding,
+	stageRawPrimaryEncoding,
+} = require('#src/resources/RecordEncoder');
 const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase');
 
 const isLMDB = process.env.HARPER_STORAGE_ENGINE === 'lmdb';
 
+function asBinary(buffer) {
+	return { ['\x10binary-data\x02']: buffer };
+}
+
 describe('PrimaryRocksDatabase metadata repair (#2012)', function () {
 	let dbPath, root, store, rawStore;
 
@@ -69,3 +80,126 @@ describe('PrimaryRocksDatabase metadata repair (#2012)', function () {
 		assert(viaGet instanceof RecordObject);
 	});
 });
+
+describe('PrimaryRocksDatabase raw-write versioning (#1762)', function () {
+	let dbPath, root, store, rawStore;
+
+	before(function () {
+		if (isLMDB) return this.skip();
+		dbPath = path.join(setupTestDBPath(), 'rocks-raw-write-versioning');
+		fs.removeSync(dbPath);
+		root = RocksDatabase.open(dbPath, {});
+		store = new PrimaryRocksDatabase(dbPath, {
+			name: 'RawWriteTest/',
+			encoder: { Encoder: RecordEncoder },
+			useVersions: true,
+			sharedStructuresKey: Symbol.for('structures'),
+		}).open();
+		store.initStore(root);
+		rawStore = RocksDatabase.open(dbPath, { name: 'RawWriteTest/', encoding: false });
+	});
+
+	after(function () {
+		rawStore?.close();
+		store?.close();
+		root?.close();
+	});
+
+	it('does not stage metadata for an encoder that is not initialized for RocksDB', function () {
+		assert.strictEqual(stageRawPrimaryEncoding({ useVersions: true, isRocksDB: false }, 123), false);
+	});
+
+	it('versions mixed-shape raw puts before classic structure id 0x42 can collide', function () {
+		for (let copy = 1; copy <= 2; copy++) {
+			store.putSync(`failure-${copy}`, {
+				backend: 'unknown',
+				method: 'generate',
+				model: 'missing',
+				success: false,
+				error_code: 'backend_not_found',
+			});
+		}
+		for (let copy = 1; copy <= 2; copy++) {
+			store.putSync(`generate-${copy}`, {
+				backend: 'deterministic',
+				method: 'generate',
+				model: 'probe',
+				success: true,
+				prompt_tokens: 1,
+				completion_tokens: 1,
+			});
+		}
+		for (let copy = 1; copy <= 2; copy++) {
+			store.putSync(`stream-${copy}`, {
+				backend: 'deterministic',
+				method: 'generateStream',
+				model: 'probe',
+				success: true,
+			});
+		}
+
+		const bytes = rawStore.getBinarySync('stream-2');
+		assert.strictEqual(bytes.readUint32BE(8), ACTION_32_BIT << 24, 'raw write must include a zero flags word');
+		assert.strictEqual(
+			bytes[12],
+			0x42,
+			`precondition: stream shape must use classic structure id 0x42; bytes=${bytes.toString('hex')}`
+		);
+		assert.deepStrictEqual(
+			{ ...store.getSync('stream-2') },
+			{ backend: 'deterministic', method: 'generateStream', model: 'probe', success: true }
+		);
+		assert(store.getEntry('stream-2').version > 0);
+	});
+
+	it('uses a positional version verbatim for sync raw puts', function () {
+		const version = 1_800_000_000_000.25;
+		store.putSync('explicit-version', { source: 'legacy-call-shape' }, version);
+		assert.strictEqual(store.getEntry('explicit-version').version, version);
+	});
+
+	it('uses a positional version verbatim for async raw puts', async function () {
+		const version = 1_800_000_000_000.5;
+		await store.put('explicit-async-version', { source: 'async-call-shape' }, version);
+		assert.strictEqual(store.getEntry('explicit-async-version').version, version);
+	});
+
+	it('replaces an unusable explicit version with a monotonic version', function () {
+		store.putSync('zero-version', { source: 'legacy-zero' }, 0);
+		assert(store.getEntry('zero-version').version > 0);
+	});
+
+	it('does not replace metadata already staged by recordUpdater', function () {
+		const stagedVersion = 1_800_000_000_001.5;
+		setNextEncoding(stagedVersion, 0);
+		store.putSync('staged-version', { source: 'record-updater' }, { version: stagedVersion + 1 });
+		assert.strictEqual(store.getEntry('staged-version').version, stagedVersion);
+	});
+
+	it('does not inherit a stale blob flag on a wrapper-staged raw write', function () {
+		const blob = createBlob(Buffer.from('prior write'));
+		blob.saveInRecord = true;
+		encodeBlobsWithFilePath(() => store.encoder.encode({ blob }), 99, root);
+		store.putSync('after-blob', { source: 'raw-write' });
+		assert.strictEqual(rawStore.getBinarySync('after-blob').readUint32BE(8), ACTION_32_BIT << 24);
+	});
+
+	it('clears wrapper metadata when a raw binary write bypasses the encoder', function () {
+		const buffer = Buffer.from([1, 2, 3]);
+		store.putSync('binary', asBinary(buffer), 123);
+		assert.deepStrictEqual(rawStore.getBinarySync('binary'), buffer);
+		store.putSync('after-binary', { source: 'next-write' });
+		assert.notStrictEqual(store.getEntry('after-binary').version, 123);
+	});
+
+	it('clears wrapper metadata when the store rejects before encoding', function () {
+		const closed = new PrimaryRocksDatabase(dbPath, {
+			name: 'ClosedRawWriteTest/',
+			encoder: { Encoder: RecordEncoder },
+		});
+		closed.initStore(root);
+		assert.throws(() => closed.putSync('closed', { source: 'failure' }, 456), /Database not open/);
+		store.putSync('after-failure', { source: 'next-write' });
+		assert.notStrictEqual(store.getEntry('after-failure').version, 456);
+	});
+});