fix: admit Maker orchestrator strategy archive batches

openspec/specs/strategy-runtime-archive-ingestion/spec.md

@@ -5,16 +5,26 @@ TBD - created by archiving change maker-archive-forwarder-conformance. Update Pu
 ## Requirements
 ### Requirement: Strategy runtime requests use a closed envelope contract
 
-The archive forwarder MUST accept a strategy runtime request only when its non-empty envelope source is `hb_runtime`, its deployment id is non-empty, it contains between one and 1000 rows, and every row targets an approved strategy runtime table.
+The archive forwarder MUST accept a strategy runtime request only when its non-empty envelope source belongs to exactly the admitted producer set `{hb_runtime, maker_orchestrator}`, its deployment id is non-empty, it contains between one and 1000 rows, and every row targets an approved strategy runtime table.
 
 The approved table set SHALL be exactly `strategy_data.policy_evaluation_events`, `strategy_data.strategy_policy_snapshots`, `strategy_data.market_identity`, `strategy_data.symbol_mapping`, and `strategy_data.inventory_settlement_events`.
 
+Conforming batches from either admitted producer MUST enter the same durable strategy-spool ownership path and return HTTP 202 without direct synchronous ClickHouse insertion.
+
 #### Scenario: All five strategy tables are submitted
 - **WHEN** one `hb_runtime` envelope contains conforming rows for all five approved tables
 - **THEN** the request MUST pass strategy contract validation
 
+#### Scenario: Maker orchestrator strategy rows are submitted
+- **WHEN** one `maker_orchestrator` envelope contains conforming rows for an approved strategy table
+- **THEN** the request MUST pass strategy contract validation
+
+#### Scenario: Either admitted producer transfers durable ownership
+- **WHEN** a conforming `hb_runtime` or `maker_orchestrator` strategy envelope is submitted
+- **THEN** the forwarder MUST durably admit it to the strategy spool and return HTTP 202 without direct ClickHouse insertion
+
 #### Scenario: Strategy and non-strategy rows are mixed
-- **WHEN** an `hb_runtime` envelope contains an approved strategy table and any non-strategy table
+- **WHEN** an envelope from either admitted producer contains an approved strategy table and any non-strategy table
 - **THEN** the entire request MUST be rejected with HTTP 400 before durable admission or ClickHouse insertion
 
 #### Scenario: Strategy table uses another source
@@ -53,10 +63,10 @@ Version `2` rows MUST contain non-empty `producer_id`, `producer_run_id`, `strea
 
 ### Requirement: Envelope and row provenance agree
 
-When a strategy row supplies `source` or `deployment_id`, its value MUST equal the corresponding envelope value. Credentials and broker configuration MUST NOT supply or override these Maker identities.
+When a strategy row supplies `source` or `deployment_id`, its value MUST equal the corresponding envelope value. Credentials and broker configuration MUST NOT supply or override these producer identities.
 
-#### Scenario: Row source differs from envelope
-- **WHEN** a row declares a source other than the envelope source
+#### Scenario: Row source differs across admitted producers
+- **WHEN** an `hb_runtime` envelope contains a row declaring `maker_orchestrator`, or a `maker_orchestrator` envelope contains a row declaring `hb_runtime`
 - **THEN** the entire request MUST be rejected before admission
 
 #### Scenario: Row deployment differs from envelope
@@ -91,9 +101,8 @@ The CEX Broker contract fixture MUST equal the pinned Maker `archive_forwarder_e
 
 ### Requirement: Non-strategy archive behavior remains compatible
 
-Requests that contain no strategy table and do not use `hb_runtime` MUST retain the existing direct ClickHouse insertion and synchronous success/failure contract.
+Requests that contain no strategy table and do not use either admitted strategy producer MUST retain the existing direct ClickHouse insertion and synchronous success/failure contract.
 
 #### Scenario: Broker market rows are submitted
 - **WHEN** a valid `broker_read` or `broker_write` request contains only supported non-strategy rows
 - **THEN** the forwarder MUST insert them through the existing direct path and MUST NOT consume strategy spool quota
-

services/archive-forwarder/strategy-contract.ts

@@ -1,6 +1,9 @@
 import type { ArchiveBatchRequest, ArchiveRow } from "./types";
 
-export const STRATEGY_ARCHIVE_SOURCE = "hb_runtime";
+export const STRATEGY_ARCHIVE_SOURCES: ReadonlySet<string> = new Set([
+	"hb_runtime",
+	"maker_orchestrator",
+]);
 
 export const STRATEGY_ARCHIVE_TABLES = [
 	"strategy_data.policy_evaluation_events",
@@ -58,7 +61,7 @@ export function classifyStrategyArchiveBatch(
 		isStrategyArchiveTable(entryTable(entry)),
 	);
 
-	if (source === STRATEGY_ARCHIVE_SOURCE) {
+	if (typeof source === "string" && STRATEGY_ARCHIVE_SOURCES.has(source)) {
 		return rows.length > 0 && rows.every((entry) =>
 			isStrategyArchiveTable(entryTable(entry)),
 		)

test/archive-forwarder-strategy-contract.test.ts

@@ -2,9 +2,11 @@ import { describe, expect, test } from "bun:test";
 import { createHash } from "node:crypto";
 import {
 	classifyStrategyArchiveBatch,
+	STRATEGY_ARCHIVE_SOURCES,
 	validateStrategyArchiveBatch,
 } from "../services/archive-forwarder/strategy-contract";
 import fixture from "./fixtures/archive_forwarder_envelope.json";
+import makerOrchestratorFixture from "./fixtures/maker_orchestrator_archive_envelope.json";
 
 const PINNED_MAKER_FIXTURE_SHA256 =
 	"5c9fd679a5a05ebce5f5158f4cc376360f24a34d9a07edeee43e94e564db3ee7";
@@ -54,6 +56,42 @@ describe("Maker strategy archive contract", () => {
 		expect(validateStrategyArchiveBatch(batch(rows))).toEqual({ ok: true });
 	});
 
+	test("classifies both admitted strategy producers", () => {
+		expect(STRATEGY_ARCHIVE_SOURCES).toEqual(
+			new Set(["hb_runtime", "maker_orchestrator"]),
+		);
+		expect(classifyStrategyArchiveBatch(fixture)).toBe("strategy");
+		expect(classifyStrategyArchiveBatch(makerOrchestratorFixture)).toBe(
+			"strategy",
+		);
+		expect(validateStrategyArchiveBatch(makerOrchestratorFixture)).toEqual({
+			ok: true,
+		});
+	});
+
+	test.each([
+		"unknown_runtime",
+		"broker_read",
+		"broker_write",
+	])("rejects %s carrying strategy rows", (source) => {
+		expect(
+			classifyStrategyArchiveBatch({
+				...makerOrchestratorFixture,
+				source,
+			}),
+		).toBe("invalid_strategy_source");
+	});
+
+	test("rejects cross-producer row and envelope provenance", () => {
+		const hbEnvelopeWithMakerRow = structuredClone(fixture);
+		hbEnvelopeWithMakerRow.rows[0].row.source = "maker_orchestrator";
+		expect(validateStrategyArchiveBatch(hbEnvelopeWithMakerRow).ok).toBe(false);
+
+		const makerEnvelopeWithHbRow = structuredClone(makerOrchestratorFixture);
+		makerEnvelopeWithHbRow.rows[0].row.source = "hb_runtime";
+		expect(validateStrategyArchiveBatch(makerEnvelopeWithHbRow).ok).toBe(false);
+	});
+
 	test.each([
 		undefined,
 		"",
@@ -104,7 +142,7 @@ describe("Maker strategy archive contract", () => {
 		expect(validateStrategyArchiveBatch(batch([])).ok).toBe(false);
 	});
 
-	test("rejects mixed tables, mixed sources, and row provenance mismatch", () => {
+	test("rejects mixed tables and row deployment mismatch", () => {
 		const strategy = v2Row("strategy_data.policy_evaluation_events", 1);
 		expect(
 			validateStrategyArchiveBatch(
@@ -114,13 +152,6 @@ describe("Maker strategy archive contract", () => {
 				]),
 			).ok,
 		).toBe(false);
-		expect(
-			classifyStrategyArchiveBatch({
-				source: "broker_write",
-				deployment_id: "maker-a",
-				rows: [strategy],
-			}),
-		).toBe("invalid_strategy_source");
 		strategy.row.deployment_id = "maker-b";
 		expect(validateStrategyArchiveBatch(batch([strategy])).ok).toBe(false);
 	});

test/archive-forwarder-strategy-request.test.ts

@@ -6,6 +6,7 @@ import {
 	type ArchiveMetricsRecorder,
 } from "../services/archive-forwarder/telemetry";
 import fixture from "./fixtures/archive_forwarder_envelope.json";
+import makerOrchestratorFixture from "./fixtures/maker_orchestrator_archive_envelope.json";
 
 const noopRecorder: ArchiveMetricsRecorder = {
 	recordCounter: () => {},
@@ -41,6 +42,29 @@ describe("strategy durable HTTP admission", () => {
 		spool.close();
 	});
 
+	test("durably admits Maker orchestrator rows without waiting for ClickHouse", async () => {
+		const spool = new StrategyArchiveSpool({ path: ":memory:" });
+		let insertCalled = false;
+		const response = await handleArchiveRequest(
+			post(makerOrchestratorFixture),
+			{
+				inserter: async () => {
+					insertCalled = true;
+					throw new Error("ClickHouse is down");
+				},
+				spool,
+				telemetry: new ArchiveForwarderTelemetry(noopRecorder),
+			},
+		);
+		expect(response.status).toBe(202);
+		expect(insertCalled).toBe(false);
+		expect(spool.stats()).toMatchObject({
+			queuedBatches: 1,
+			queuedWork: makerOrchestratorFixture.rows.length,
+		});
+		spool.close();
+	});
+
 	test("keeps broker traffic on direct synchronous insertion", async () => {
 		const spool = new StrategyArchiveSpool({ path: ":memory:" });
 		let insertCalled = false;

test/fixtures/maker_orchestrator_archive_envelope.json

@@ -0,0 +1,35 @@
+{
+  "source": "maker_orchestrator",
+  "deployment_id": "maker-deployment",
+  "rows": [
+    {
+      "table": "strategy_data.inventory_settlement_events",
+      "row": {
+        "source": "maker_orchestrator",
+        "deployment_id": "maker-deployment",
+        "schema_version": "2",
+        "producer_id": "maker_orchestrator:maker-deployment:orchestrator",
+        "producer_run_id": "run-1",
+        "controller_id": "orchestrator",
+        "controller_type": "maker_orchestrator",
+        "connector_name": "",
+        "exchange": "",
+        "trading_pair": "",
+        "market_id": "pool-1",
+        "run_id": "run-1",
+        "seq": 1,
+        "stream_name": "maker_runtime_state",
+        "stream_seq": 1,
+        "archive_event_id": "run-1:maker_runtime_state:1",
+        "event_time_ms": 1719999999997,
+        "emitted_at_ms": 1720000000000,
+        "event_kind": "orchestrator_retained_state",
+        "token": "ARB",
+        "account": "wallet",
+        "reservation_id": "reservation-1",
+        "workflow_state": "published",
+        "payload_json": "{\"pool_id\":\"pool-1\"}"
+      }
+    }
+  ]
+}