From 26d8421ef780bb5ba3a02f6645734ecedc5fe281 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 2 Jul 2026 20:51:00 +0800 Subject: [PATCH] feat(skills): add governed reply router --- skills/reply-router/SKILL.md | 109 ++++++ skills/reply-router/X.yaml | 164 +++++++++ .../reply-router/fixtures/harness-result.json | 14 + .../sealed-unsubscribe-suppression.yaml | 18 + .../fixtures/stop-ambiguous-or-unsealed.yaml | 17 + .../tools/data/local/manifest.json | 77 ++++ skills/reply-router/tools/data/local/run.mjs | 335 ++++++++++++++++++ 7 files changed, 734 insertions(+) create mode 100644 skills/reply-router/SKILL.md create mode 100644 skills/reply-router/X.yaml create mode 100644 skills/reply-router/fixtures/harness-result.json create mode 100644 skills/reply-router/fixtures/sealed-unsubscribe-suppression.yaml create mode 100644 skills/reply-router/fixtures/stop-ambiguous-or-unsealed.yaml create mode 100644 skills/reply-router/tools/data/local/manifest.json create mode 100644 skills/reply-router/tools/data/local/run.mjs diff --git a/skills/reply-router/SKILL.md b/skills/reply-router/SKILL.md new file mode 100644 index 000000000..7a999e1ab --- /dev/null +++ b/skills/reply-router/SKILL.md @@ -0,0 +1,109 @@ +--- +name: reply-router +description: Classify an inbound reply against a sealed original send receipt and either append a recipient-keyed suppression event to a hosted data-store or emit a bounded governed routing decision. The skill never sends mail. +runx: + category: ops +--- + +# Reply Router + +`reply-router` reads one inbound reply together with the sealed original send +receipt that produced it, classifies the reply against a suppression policy, and +branches. The skill is a read-and-route judgment: it never sends mail, never +mints a new send, and never edits the original send receipt. The actual send is +a separate governed `send-as` run a downstream driver issues by name. + +## What This Skill Does + +The skill reads three pieces of evidence: + +- `inbound_reply{content, received_from, received_at}` +- `original_send_receipt{send_plan, principal, receipt_id, checksum}` +- `suppression_policy{unsubscribe_signals, confidence_threshold}` + +It validates that the original send receipt is sealed (the receipt envelope is +present and the checksum matches the named `send_plan`) and that the policy +names at least one unsubscribe signal. It then classifies the reply by +searching the content for any of the policy's `unsubscribe_signals` strings or +patterns. A high-confidence unsubscribe match appends a suppression event to +the hosted data-store `registry:runx/data-store@0.1.2` via an `append_event` +with an idempotency key and an expected_version CAS. The durable record is the +compliance block the next `send-as` preflight reads as a fail-closed gate. + +For other classifications (`interested`, `objection`, `out-of-office`, +`wrong-person`), the skill emits a typed `runx.reply.routing.v1` decision +naming a bounded `send_target` and a `principal`. A downstream `send-as` run +honors that decision later; the skill itself does not dispatch. + +For an unsealed or missing original send receipt, or for an inbound reply +whose content does not match any policy signal and where the agent cannot +ground a typed classification, the skill escalates to a human approval lane +and emits no suppression event and no routing decision. + +## When To Use It + +- An operator has a sealed original send receipt and needs a typed decision + on how to handle the inbound reply before any subsequent send. +- A workflow needs to prove a recipient was unsubscribed via a durable + data-store record that the next `send-as` preflight can verify. +- A run should keep suppression and routing out of the actual mail path. + +## When Not To Use It + +- To actually send, queue, schedule, or otherwise move a message. Use a + separate governed `send-as` run for that effect. +- To append a suppression event without a sealed original send receipt. +- To emit a routing decision when the inbound content is ambiguous or the + reply cannot be grounded in a policy signal. +- To invent a classification, an unsubscribe match, or a routing target that + the input evidence does not support. + +## Procedure + +1. Read `inbound_reply`, `original_send_receipt`, and `suppression_policy`. + Reject any missing or unclear top-level object. +2. Verify the `original_send_receipt` is sealed: `receipt_id` and `checksum` + are present, `send_plan` and `principal` are non-empty strings, and + `checksum` matches the named `send_plan`. If any of these is missing or + mismatched, escalate to the human approval lane and emit no outputs. +3. Verify `suppression_policy.unsubscribe_signals` is a non-empty list of + strings. The `confidence_threshold` must be a number in (0, 1]. +4. Search `inbound_reply.content` for each policy `unsubscribe_signals` + string using a case-insensitive match. If at least one signal is found + and the match is exact (not negated, not quoted as not-an-unsubscribe), + classify the reply as `unsubscribe` with confidence 1.0 and evidence + naming the matched signal and the offset in the reply content. +5. For an `unsubscribe` classification, append a suppression event to the + hosted data-store via `append_event` with: + - `aggregate_id` = the reply's `received_from` (the recipient key) + - `idempotency_key` = sha256 of `{receipt_id}:{received_from}:{signal}` + - `expected_version` = the value from a prior `read_projection` against + `aggregate_id` (0 if no prior projection exists) + - The event payload names the matched signal, the original receipt id, + and the inbound reply received_at timestamp. + The skill emits a `suppression_result{aggregate_id, idempotency_key, + before_version, after_version}`. The skill does NOT emit a routing + decision on this path. +6. If no unsubscribe signal matches, attempt to ground a typed + classification from a bounded set (`interested`, `objection`, + `out-of-office`, `wrong-person`) using the reply content and the + original send plan. If grounding succeeds with confidence above + `confidence_threshold`, emit `runx.reply.routing.v1{classification, + send_target, principal}` naming a bounded send target the downstream + `send-as` run honors. The skill consumes nothing and does not dispatch. +7. If no typed classification can be grounded above + `confidence_threshold`, escalate to the human approval lane and emit + no outputs. +8. Never invent a match, classification, or aggregate version. Never append + a suppression event without a sealed original send receipt. Never + classify a reply whose `received_from` does not match the principal of + the original send receipt. + +## Outputs + +- `classification{type, confidence, evidence}` for every sealed run. +- `suppression_result{aggregate_id, idempotency_key, before_version, + after_version}` when the reply is an unsubscribe. +- `runx.reply.routing.v1{classification, send_target, principal}` when the + reply is routed. +- `escalation_reason` when the run stops for a human approval lane. diff --git a/skills/reply-router/X.yaml b/skills/reply-router/X.yaml new file mode 100644 index 000000000..c8127b46d --- /dev/null +++ b/skills/reply-router/X.yaml @@ -0,0 +1,164 @@ +skill: reply-router +version: "0.1.0" + +catalog: + kind: graph + audience: public + visibility: public + role: context + +harness: + cases: + - name: sealed_unsubscribe_suppression + runner: reply_route + inputs: + inbound_reply: + content: "Please unsubscribe me from this list. No more emails please." + received_from: "ops@example.com" + received_at: "2026-07-02T10:00:00Z" + original_send_receipt: + send_plan: '{"to":"ops@example.com","subject":"weekly digest","from":"ops@example.com","body_digest":"weekly news"}' + principal: "ops" + receipt_id: "runx:receipt:sha256:abc123replyrouter" + checksum: "sha256:49e1b30d46a6eee6156bcf9a1e2d8400f17d69a3e271d541b61a7ad49c7c05ad" + suppression_policy: + unsubscribe_signals: + - unsubscribe + - opt out + - remove me + confidence_threshold: 0.8 + operator_context: Append a recipient-keyed suppression event to the data-store via ungated CAS. No routing decision is emitted on this path; the next send-as preflight must read the durable record as a fail-closed block. + caller: + answers: + agent_task.reply-classify.output: + classification: + type: unsubscribe + confidence: 1.0 + evidence: matched policy signal "unsubscribe" at offset 7 + suppression_result: + aggregate_id: ops@example.com + idempotency_key: bbdc596dd2d71413bc2bce898a98a2a52d32ed5d86abd73c185a2ab3a3eed733 + before_version: 0 + after_version: 1 + suppression_event: + aggregate_id: ops@example.com + idempotency_key: bbdc596dd2d71413bc2bce898a98a2a52d32ed5d86abd73c185a2ab3a3eed733 + event: + type: reply.unsubscribe_recorded + payload: + matched_signal: unsubscribe + match_offset: 7 + original_receipt_id: runx:receipt:sha256:abc123replyrouter + received_at: "2026-07-02T10:00:00Z" + data_store_ref: registry:runx/data-store@0.1.2 + routing_decision: null + escalation_reason: null + expect: + status: sealed + receipt: + schema: runx.receipt.v1 + state: sealed + - name: stop_ambiguous_or_unsealed + runner: reply_route + inputs: + inbound_reply: + content: "ok thanks" + received_from: "ops@example.com" + received_at: "2026-07-02T10:00:00Z" + original_send_receipt: + send_plan: '{"to":"ops@example.com","subject":"weekly digest","from":"ops@example.com","body_digest":"weekly news"}' + principal: "ops" + receipt_id: "runx:receipt:sha256:abc123replyrouter" + checksum: "sha256:49e1b30d46a6eee6156bcf9a1e2d8400f17d69a3e271d541b61a7ad49c7c05ad" + suppression_policy: + unsubscribe_signals: + - unsubscribe + - opt out + - remove me + confidence_threshold: 0.8 + operator_context: Ambiguous reply with no policy match and no grounded typed classification; the classify sub-step must block to needs_human with no suppression write and no routing decision. + expect: + status: needs_agent + +runners: + reply_route: + default: true + type: graph + inputs: + inbound_reply: + type: json + required: true + description: Inbound reply content, recipient, and timestamp. + original_send_receipt: + type: json + required: true + description: Sealed original send receipt with send_plan, principal, receipt_id, and checksum. + suppression_policy: + type: json + required: true + description: Unsubscribe signal list and confidence threshold. + graph: + name: reply-router-read-classify-and-persist + steps: + - id: read_suppression + label: read projection using the registry:runx/data-store@0.1.2 data-source contract + tool: data.source + scopes: + - runx:data:read + inputs: + operation: read_projection + data_source_ref: local://reply-router/suppressions + store_id: reply-router-suppressions-v1 + resource: suppression_events + aggregate_id: "$input.inbound_reply.received_from" + - id: classify + run: + type: agent-task + agent: analyst + task: reply-classify + outputs: + classification: object + suppression_result: object + suppression_event: object + routing_decision: object + escalation_reason: string + inputs: + inbound_reply: "$input.inbound_reply" + original_send_receipt: "$input.original_send_receipt" + suppression_policy: "$input.suppression_policy" + context: + current_projection: read_suppression.data_operation_result.data.projection + instructions: Classify only from the supplied reply, sealed receipt, policy, and current projection. Never invent intent. An unsubscribe result must emit a suppression event and no routing decision; ambiguous or unsealed input must require human review. + - id: append_suppression + when: + field: classify.classification.type + equals: unsubscribe + label: append through the registry:runx/data-store@0.1.2 data-source contract + tool: data.source + scopes: + - runx:data:append + inputs: + operation: append_event + data_source_ref: local://reply-router/suppressions + store_id: reply-router-suppressions-v1 + resource: suppression_events + context: + aggregate_id: classify.suppression_event.aggregate_id + expected_version: read_suppression.data_operation_result.data.projection.version + idempotency_key: classify.suppression_event.idempotency_key + event: classify.suppression_event.event + - id: readback_suppression + when: + field: classify.classification.type + equals: unsubscribe + label: verify through the registry:runx/data-store@0.1.2 data-source contract + tool: data.source + scopes: + - runx:data:read + inputs: + operation: read_projection + data_source_ref: local://reply-router/suppressions + store_id: reply-router-suppressions-v1 + resource: suppression_events + context: + aggregate_id: classify.suppression_event.aggregate_id diff --git a/skills/reply-router/fixtures/harness-result.json b/skills/reply-router/fixtures/harness-result.json new file mode 100644 index 000000000..1c015203e --- /dev/null +++ b/skills/reply-router/fixtures/harness-result.json @@ -0,0 +1,14 @@ +{ + "status": "passed", + "case_count": 2, + "assertion_error_count": 0, + "assertion_errors": [], + "case_names": [ + "sealed_unsubscribe_suppression", + "stop_ambiguous_or_unsealed" + ], + "receipt_ids": [ + "sha256:a21bde8ef82146536fea77aeedba6f0e62e29a727d5315e409abc8bfc0e74425" + ], + "graph_case_count": 2 +} diff --git a/skills/reply-router/fixtures/sealed-unsubscribe-suppression.yaml b/skills/reply-router/fixtures/sealed-unsubscribe-suppression.yaml new file mode 100644 index 000000000..78c5c0969 --- /dev/null +++ b/skills/reply-router/fixtures/sealed-unsubscribe-suppression.yaml @@ -0,0 +1,18 @@ +runner: reply_route +inputs: + inbound_reply: + content: "Please unsubscribe me from this list. No more emails please." + received_from: "ops@example.com" + received_at: "2026-07-02T10:00:00Z" + original_send_receipt: + send_plan: '{"to":"ops@example.com","subject":"weekly digest","from":"ops@example.com","body_digest":"weekly news"}' + principal: "ops" + receipt_id: "runx:receipt:sha256:abc123replyrouter" + checksum: "sha256:49e1b30d46a6eee6156bcf9a1e2d8400f17d69a3e271d541b61a7ad49c7c05ad" + suppression_policy: + unsubscribe_signals: + - unsubscribe + - opt out + - remove me + confidence_threshold: 0.8 + diff --git a/skills/reply-router/fixtures/stop-ambiguous-or-unsealed.yaml b/skills/reply-router/fixtures/stop-ambiguous-or-unsealed.yaml new file mode 100644 index 000000000..16a5f2163 --- /dev/null +++ b/skills/reply-router/fixtures/stop-ambiguous-or-unsealed.yaml @@ -0,0 +1,17 @@ +runner: reply_route +inputs: + inbound_reply: + content: "ok thanks" + received_from: "ops@example.com" + received_at: "2026-07-02T10:00:00Z" + original_send_receipt: + send_plan: '{"to":"ops@example.com","subject":"weekly digest","from":"ops@example.com","body_digest":"weekly news"}' + principal: "ops" + receipt_id: "runx:receipt:sha256:abc123replyrouter" + checksum: "sha256:49e1b30d46a6eee6156bcf9a1e2d8400f17d69a3e271d541b61a7ad49c7c05ad" + suppression_policy: + unsubscribe_signals: + - unsubscribe + - opt out + - remove me + confidence_threshold: 0.8 diff --git a/skills/reply-router/tools/data/local/manifest.json b/skills/reply-router/tools/data/local/manifest.json new file mode 100644 index 000000000..ebf37ea0f --- /dev/null +++ b/skills/reply-router/tools/data/local/manifest.json @@ -0,0 +1,77 @@ +{ + "schema": "runx.tool.manifest.v1", + "name": "data.local", + "version": "0.1.0", + "description": "Local JSON event-store adapter for the provider-agnostic runx data operation envelope.", + "source": { + "type": "cli-tool", + "command": "node", + "args": ["./run.mjs"] + }, + "inputs": { + "operation": { + "type": "string", + "required": true, + "description": "append_event, read_events, or read_projection." + }, + "data_source_ref": { + "type": "string", + "required": true, + "description": "Stable logical data-source reference." + }, + "store_id": { + "type": "string", + "required": false, + "description": "Local fixture store id. Provider adapters may ignore this." + }, + "resource": { + "type": "string", + "required": true, + "description": "Declared resource, stream, table, keyspace, or projection name." + }, + "aggregate_id": { + "type": "string", + "required": true, + "description": "Stream or partition key." + }, + "expected_version": { + "type": "number", + "required": false, + "description": "Required current stream version for append_event." + }, + "idempotency_key": { + "type": "string", + "required": false, + "description": "Stable retry key for append_event." + }, + "event": { + "type": "json", + "required": false, + "description": "Domain event or transition packet for append_event." + }, + "limit": { + "type": "number", + "required": false, + "default": 50, + "description": "Maximum events returned by read_events." + } + }, + "scopes": ["runx:data:read", "runx:data:append"], + "runx": { + "artifacts": { + "named_emits": { + "data_operation_result": "runx.data.operation_result.v1" + }, + "wrap_as": "data_operation_result" + } + }, + "runtime": { + "command": "node", + "args": ["./run.mjs"] + }, + "output": { + "packet": "runx.data.operation_result.v1", + "wrap_as": "data_operation_result" + }, + "toolkit_version": "0.1.4" +} diff --git a/skills/reply-router/tools/data/local/run.mjs b/skills/reply-router/tools/data/local/run.mjs new file mode 100644 index 000000000..659c9702b --- /dev/null +++ b/skills/reply-router/tools/data/local/run.mjs @@ -0,0 +1,335 @@ +import crypto from "node:crypto"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const SCHEMA = "runx.data.operation_result.v1"; +const PROVIDER = "local-json-event-store"; + +const inputs = readInputs(); +const operation = stringInput("operation"); + +let result; +if (operation === "append_event") { + result = appendEvent(inputs); +} else if (operation === "read_events") { + result = readEvents(inputs); +} else if (operation === "read_projection") { + result = readProjection(inputs); +} else { + throw new Error("operation must be append_event, read_events, or read_projection"); +} + +process.stdout.write(`${JSON.stringify(result, null, 2)}\n`); + +function readInputs() { + const raw = process.env.RUNX_INPUTS_PATH + ? fs.readFileSync(process.env.RUNX_INPUTS_PATH, "utf8") + : process.env.RUNX_INPUTS_JSON || "{}"; + return JSON.parse(raw); +} + +function appendEvent(rawInputs) { + const envelope = baseEnvelope(rawInputs, "append_event"); + const expectedVersion = numberInput("expected_version"); + const idempotencyKey = stringInput("idempotency_key"); + const event = objectInput("event"); + const store = readStore(rawInputs); + const stream = streamFor(store, envelope.resource, envelope.aggregate_id); + const eventDigest = sha256Json(event); + const existing = stream.events.find((entry) => entry.idempotency_key === idempotencyKey); + + if (existing) { + if (existing.event_digest !== eventDigest) { + return conflictResult(envelope, stream, { + idempotency_key: idempotencyKey, + event_digest: eventDigest, + reason: "idempotency key was reused with different event content", + provider_evidence: providerEvidence(store, envelope), + }); + } + return { + ...envelope, + status: "idempotent_replay", + before_version: stream.version, + after_version: stream.version, + idempotency_key: idempotencyKey, + event_ref: existing.event_ref, + event_digest: existing.event_digest, + result_digest: sha256Json(existing), + projection_digest: projectionDigest(stream), + events: [], + rows: [], + redactions: [], + stop_conditions: [], + provider_evidence: providerEvidence(store, envelope), + }; + } + + if (stream.version !== expectedVersion) { + return conflictResult(envelope, stream, { + idempotency_key: idempotencyKey, + event_digest: eventDigest, + reason: `expected version ${expectedVersion}, got ${stream.version}`, + provider_evidence: providerEvidence(store, envelope), + }); + } + + const nextVersion = stream.version + 1; + const eventRef = `${envelope.resource}:${envelope.aggregate_id}:${nextVersion}`; + const record = { + event_ref: eventRef, + version: nextVersion, + event_type: eventType(event), + event, + event_digest: eventDigest, + idempotency_key: idempotencyKey, + committed_at: typeof rawInputs.observed_at === "string" ? rawInputs.observed_at : "1970-01-01T00:00:00.000Z", + }; + stream.events.push(record); + stream.version = nextVersion; + writeStore(rawInputs, store); + + return { + ...envelope, + status: "committed", + before_version: expectedVersion, + after_version: nextVersion, + idempotency_key: idempotencyKey, + event_ref: eventRef, + event_digest: eventDigest, + result_digest: sha256Json(record), + projection_digest: projectionDigest(stream), + events: [], + rows: [], + redactions: [], + stop_conditions: [], + provider_evidence: providerEvidence(store, envelope), + }; +} + +function conflictResult(envelope, stream, { idempotency_key, event_digest, reason, provider_evidence }) { + const stop = { + code: "conflict", + message: reason, + }; + return { + ...envelope, + status: "conflict", + before_version: stream.version, + after_version: stream.version, + idempotency_key, + event_ref: null, + event_digest, + result_digest: sha256Json(stop), + projection_digest: projectionDigest(stream), + events: [], + rows: [], + redactions: [], + stop_conditions: [stop], + provider_evidence, + }; +} + +function readEvents(rawInputs) { + const envelope = baseEnvelope(rawInputs, "read_events"); + const limit = boundedLimit(rawInputs.limit); + const store = readStore(rawInputs); + const stream = streamFor(store, envelope.resource, envelope.aggregate_id); + const events = stream.events.slice(Math.max(0, stream.events.length - limit)); + return { + ...envelope, + status: "read", + before_version: stream.version, + after_version: stream.version, + idempotency_key: null, + event_ref: null, + event_digest: null, + result_digest: sha256Json(events), + projection_digest: projectionDigest(stream), + events, + rows: events, + redactions: [], + stop_conditions: [], + provider_evidence: providerEvidence(store, envelope), + }; +} + +function readProjection(rawInputs) { + const envelope = baseEnvelope(rawInputs, "read_projection"); + const store = readStore(rawInputs); + const stream = streamFor(store, envelope.resource, envelope.aggregate_id); + const projection = { + aggregate_id: envelope.aggregate_id, + resource: envelope.resource, + version: stream.version, + event_count: stream.events.length, + last_event_ref: stream.events.at(-1)?.event_ref ?? null, + last_event_type: stream.events.at(-1)?.event_type ?? null, + event_digests: stream.events.map((entry) => entry.event_digest), + }; + return { + ...envelope, + status: "read", + before_version: stream.version, + after_version: stream.version, + idempotency_key: null, + event_ref: null, + event_digest: null, + result_digest: sha256Json(projection), + projection_digest: sha256Json(projection), + projection, + events: [], + rows: [], + redactions: [], + stop_conditions: [], + provider_evidence: providerEvidence(store, envelope), + }; +} + +function baseEnvelope(rawInputs, operation) { + return { + schema: SCHEMA, + data_source_ref: stringInput("data_source_ref"), + provider: PROVIDER, + operation, + resource: safeName(stringInput("resource"), "resource"), + aggregate_id: safeName(stringInput("aggregate_id"), "aggregate_id"), + }; +} + +function streamFor(store, resource, aggregateId) { + store.resources[resource] ??= { streams: {} }; + store.resources[resource].streams[aggregateId] ??= { version: 0, events: [] }; + return store.resources[resource].streams[aggregateId]; +} + +function readStore(rawInputs) { + const file = storePath(rawInputs); + if (!fs.existsSync(file)) { + return { + schema: "runx.local_data_store.v1", + store_id: localStoreId(rawInputs), + resources: {}, + }; + } + const parsed = JSON.parse(fs.readFileSync(file, "utf8")); + if (!parsed || typeof parsed !== "object" || parsed.schema !== "runx.local_data_store.v1") { + throw new Error("local data store file has an invalid schema"); + } + parsed.resources ??= {}; + return parsed; +} + +function writeStore(rawInputs, store) { + const file = storePath(rawInputs); + fs.mkdirSync(path.dirname(file), { recursive: true }); + const tmp = `${file}.${process.pid}.tmp`; + fs.writeFileSync(tmp, `${JSON.stringify(store, null, 2)}\n`); + fs.renameSync(tmp, file); +} + +function storePath(rawInputs) { + const storeId = localStoreId(rawInputs); + return path.join(os.tmpdir(), "runx-data-store", `${storeId}.json`); +} + +function localStoreId(rawInputs) { + if (typeof rawInputs.store_id === "string" && rawInputs.store_id.trim().length > 0) { + return safeName(rawInputs.store_id, "store_id"); + } + const ref = typeof rawInputs.data_source_ref === "string" && rawInputs.data_source_ref.length > 0 + ? rawInputs.data_source_ref + : "default"; + return `source-${crypto.createHash("sha256").update(ref).digest("hex").slice(0, 24)}`; +} + +function providerEvidence(store, envelope) { + return { + provider: PROVIDER, + store_id: store.store_id, + resource: envelope.resource, + aggregate_id: envelope.aggregate_id, + storage_class: "local-fixture", + }; +} + +function projectionDigest(stream) { + return sha256Json({ + version: stream.version, + event_digests: stream.events.map((entry) => entry.event_digest), + }); +} + +function readValue(name) { + return inputs[name]; +} + +function stringInput(name) { + const value = readValue(name); + if (typeof value !== "string" || value.trim().length === 0) { + throw new Error(`${name} is required`); + } + return value.trim(); +} + +function numberInput(name) { + const value = readValue(name); + if (!Number.isInteger(value) || value < 0) { + throw new Error(`${name} must be a non-negative integer`); + } + return value; +} + +function objectInput(name) { + const value = readValue(name); + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new Error(`${name} must be an object`); + } + return value; +} + +function eventType(event) { + const explicit = safeEventToken(event.type) ?? safeEventToken(event.event_type); + if (explicit) return explicit; + const family = safeEventToken(event.effect_family); + const operation = safeEventToken(event.operation); + if (family && operation) return `${family}.${operation}`; + if (operation) return operation; + return "data.event"; +} + +function safeEventToken(value) { + if (typeof value !== "string") return undefined; + const text = value.trim(); + return /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(text) ? text : undefined; +} + +function boundedLimit(value) { + if (value === undefined || value === null) return 50; + if (!Number.isInteger(value) || value < 1 || value > 500) { + throw new Error("limit must be an integer from 1 to 500"); + } + return value; +} + +function safeName(value, field) { + const text = String(value || "").trim(); + const pattern = field === "aggregate_id" + ? /^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,191}$/ + : /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; + if (!pattern.test(text)) { + throw new Error(`${field} must be a safe identifier`); + } + return text; +} + +function sha256Json(value) { + return `sha256:${crypto.createHash("sha256").update(canonicalJson(value)).digest("hex")}`; +} + +function canonicalJson(value) { + if (value === null || typeof value !== "object") return JSON.stringify(value); + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; + return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${canonicalJson(value[key])}`).join(",")}}`; +}