Complete authenticated Epiphany operator bridge
This commit is contained in:
Meta Vens
parent
82dd058d42
commit
e0e612c69c
17 changed files with 383 additions and 39 deletions
|
|
@ -4,17 +4,35 @@ export const operatorAdmissionType="bifrost.operator_command.delivery";
|
|||
export const operatorAdmissionSchema="bifrost.operator_command.delivery.v0";
|
||||
export const operatorResultType="epiphany.operator_command.result";
|
||||
export const operatorResultSchema="epiphany.operator_command.result.v0";
|
||||
export const operatorResultReceiptType="epiphany.operator_command.sealed_result";
|
||||
export const operatorResultReceiptSchema="epiphany.operator_command.sealed_result.v0";
|
||||
export const operatorTerminalType="bifrost.operator_command.terminal";
|
||||
export const operatorTerminalSchema="bifrost.operator_command.terminal.v0";
|
||||
export const operatorDiscordDeliveryType="bifrost.discord.epiphany_operator_delivery";
|
||||
export const operatorDiscordDeliverySchema="bifrost.discord.epiphany_operator_delivery.v0";
|
||||
|
||||
export function buildOperatorDefinitions(defineDocumentType){const make=(type,schemaId,name,parse)=>defineDocumentType({type,schemaName:type,schemaId,schemaVersion:schemaId,contentHash:schemaId,global:false,name,schema:{parse}});return {request:make(operatorRequestType,operatorRequestSchema,"requestId",parseOperatorRequest),admission:make(operatorAdmissionType,operatorAdmissionSchema,"admissionId",parseOperatorAdmission),result:make(operatorResultType,operatorResultSchema,"resultId",parseOperatorResult)};}
|
||||
export function buildOperatorDefinitions(defineDocumentType){const make=(type,schemaId,name,parse)=>defineDocumentType({type,schemaName:type,schemaId,schemaVersion:schemaId,contentHash:schemaId,global:false,name,schema:{parse}});return {request:make(operatorRequestType,operatorRequestSchema,"requestId",parseOperatorRequest),admission:make(operatorAdmissionType,operatorAdmissionSchema,"admissionId",parseOperatorAdmission),result:make(operatorResultType,operatorResultSchema,"resultId",parseOperatorResult),sealedResult:make(operatorResultReceiptType,operatorResultReceiptSchema,"commandId",parseOperatorResultReceipt),terminal:make(operatorTerminalType,operatorTerminalSchema,"requestId",parseOperatorTerminal),discordDelivery:make(operatorDiscordDeliveryType,operatorDiscordDeliverySchema,"deliveryId",parseOperatorDiscordDelivery)};}
|
||||
export function parseOperatorRequest(value){const v=exactRecord(value,["schemaName","schemaVersion","requestId","commandId","nonce","sourceEventId","sourceActorDiscordId","discordGuildId","discordChannelId","discordMessageId","targetRuntimeId","issuedAt","expiresAt","producerId","producerRuntimeId","authorityClass","status","command"],"operator request"); exact(v.schemaName,operatorRequestType,"schemaName");exact(v.schemaVersion,operatorRequestSchema,"schemaVersion"); for(const k of ["requestId","commandId","nonce","sourceEventId","sourceActorDiscordId","discordGuildId","discordChannelId","discordMessageId","targetRuntimeId","producerRuntimeId"])text(v[k],k); exact(v.producerId,"voidbot","producerId");exact(v.authorityClass,"operator_request_only","authorityClass");exact(v.status,"pending","status"); const issued=date(v.issuedAt,"issuedAt"),expires=date(v.expiresAt,"expiresAt");if(expires<=issued||expires-issued>60000)throw new Error("operator request TTL must be positive and at most 60 seconds.");parseCommand(v.command);return v;}
|
||||
export function parseOperatorAdmission(v){v=exactRecord(v,["schemaName","schemaVersion","admissionId","packet","packetSha256","sourceObserverId","sourceObserverRuntimeId","provider","bifrostAdmissionReceiptId","authority","providerIdentityId","providerSignature"],"operator admission");exact(v.schemaName,operatorAdmissionType,"schemaName");exact(v.schemaVersion,operatorAdmissionSchema,"schemaVersion");for(const k of ["admissionId","packetSha256","sourceObserverRuntimeId","bifrostAdmissionReceiptId","providerIdentityId"])text(v[k],k);exact(v.sourceObserverId,"voidbot","sourceObserverId");exact(v.provider,"bifrost","provider");exact(v.authority,"exact_operator_command_only","authority");if(!(v.providerSignature instanceof Uint8Array)||v.providerSignature.length!==64)throw new Error("providerSignature must be 64 bytes.");parsePacket(v.packet);return v;}
|
||||
export function parseOperatorAdmission(v){v=exactRecord(v,["schemaName","schemaVersion","admissionId","packet","packetSha256","sourceObserverId","sourceObserverRuntimeId","provider","bifrostAdmissionReceiptId","authority","providerIdentityId","providerSignature"],"operator admission");exact(v.schemaName,operatorAdmissionType,"schemaName");exact(v.schemaVersion,operatorAdmissionSchema,"schemaVersion");for(const k of ["admissionId","packetSha256","sourceObserverRuntimeId","bifrostAdmissionReceiptId","providerIdentityId"])text(v[k],k);exact(v.sourceObserverId,"voidbot","sourceObserverId");exact(v.provider,"bifrost","provider");exact(v.authority,"exact_operator_command_only","authority");v.providerSignature=bytes(v.providerSignature,"providerSignature",64);parsePacket(v.packet);return v;}
|
||||
export function parseOperatorResult(v){v=exactRecord(v,["schemaVersion","resultId","commandId","packetSha256","targetRuntimeId","disposition","consequenceKind","consequenceRef","completedAt","privateStateExposed","operatorStatus","stateStatus","coordinatorAction","brakeStatus","detail"],"operator result");exact(v.schemaVersion,operatorResultSchema,"schemaVersion");for(const k of ["resultId","commandId","packetSha256","targetRuntimeId","disposition","consequenceKind","completedAt"])text(v[k],k);if(!["observed","applied","refused"].includes(v.disposition))throw new Error("result disposition is invalid.");if(v.privateStateExposed!==false)throw new Error("operator result must not expose private state.");for(const k of ["consequenceRef","operatorStatus","stateStatus","coordinatorAction","brakeStatus","detail"])if(typeof v[k]!=="string")throw new Error(`${k} must be a string.`);date(v.completedAt,"completedAt");return v;}
|
||||
export function parseOperatorResultReceipt(v){v=exactRecord(v,["schemaVersion","result","resultPayloadSha256","commandId","packetSha256","targetRuntimeId","completedAt","providerIdentityId","privateStateExposed","executorSignature"],"operator sealed result receipt");exact(v.schemaVersion,operatorResultReceiptSchema,"schemaVersion");parseOperatorResult(v.result);for(const k of ["resultPayloadSha256","commandId","packetSha256","targetRuntimeId","completedAt","providerIdentityId"])text(v[k],k);date(v.completedAt,"completedAt");if(v.privateStateExposed!==false)throw new Error("operator sealed result receipt must not expose private state.");v.executorSignature=bytes(v.executorSignature,"executorSignature",64);return v;}
|
||||
export function parseOperatorTerminal(v){v=exactRecord(v,["schemaVersion","requestId","commandId","status","failureCode","recordedAt","retryable","privateStateExposed"],"operator terminal disposition");exact(v.schemaVersion,operatorTerminalSchema,"schemaVersion");for(const k of ["requestId","commandId","failureCode","recordedAt"])text(v[k],k);exact(v.status,"refused","status");date(v.recordedAt,"recordedAt");if(v.retryable!==false||v.privateStateExposed!==false)throw new Error("operator terminal disposition must be non-retryable and private-state-free.");return v;}
|
||||
export function parseOperatorDiscordDelivery(v){v=exactRecord(v,["schemaVersion","deliveryId","requestId","requestPayloadSha256","commandId","discordGuildId","discordChannelId","discordInteractionId","targetRuntimeId","status","disposition","failureCode","detail","operatorStatus","stateStatus","coordinatorAction","brakeStatus","sealedResultPayloadSha256","executorSignatureSha256","resultProviderIdentityId","recordedAt","privateStateExposed"],"operator Discord delivery");exact(v.schemaVersion,operatorDiscordDeliverySchema,"schemaVersion");for(const k of ["deliveryId","requestId","requestPayloadSha256","commandId","discordGuildId","discordChannelId","discordInteractionId","targetRuntimeId","status","recordedAt"])text(v[k],k);if(!/^sha256-[0-9a-f]{64}$/.test(v.requestPayloadSha256))throw new Error("operator Discord delivery request digest is invalid.");if(!["completed","refused"].includes(v.status))throw new Error("operator Discord delivery status is invalid.");for(const k of ["disposition","failureCode","detail","operatorStatus","stateStatus","coordinatorAction","brakeStatus","sealedResultPayloadSha256","executorSignatureSha256","resultProviderIdentityId"])if(typeof v[k]!=="string")throw new Error(`${k} must be a string.`);for(const k of ["detail","operatorStatus","stateStatus","coordinatorAction","brakeStatus"])if(v[k].length>512)throw new Error(`operator Discord delivery ${k} exceeds 512 characters.`);if(v.status==="completed"&&(!v.disposition||!/^sha256-[0-9a-f]{64}$/.test(v.sealedResultPayloadSha256)||!/^sha256-[0-9a-f]{64}$/.test(v.executorSignatureSha256)||!v.resultProviderIdentityId))throw new Error("completed operator Discord delivery lacks sealed-result binding.");if(v.status==="refused"&&!v.failureCode)throw new Error("refused operator Discord delivery lacks failure code.");date(v.recordedAt,"recordedAt");if(v.privateStateExposed!==false)throw new Error("operator Discord delivery must not expose private state.");return v;}
|
||||
export function operatorRequestTuple(v){return [v.schemaName,v.schemaVersion,v.requestId,v.commandId,v.nonce,v.sourceEventId,v.sourceActorDiscordId,v.discordGuildId,v.discordChannelId,v.discordMessageId,v.targetRuntimeId,v.issuedAt,v.expiresAt,v.producerId,v.producerRuntimeId,v.authorityClass,v.status,commandTuple(v.command)];}
|
||||
export function packetTuple(p){return [p.commandId,p.nonce,p.sourceEventId,p.sourceActorId,p.discordGuildId,p.discordChannelId,p.discordMessageId,p.targetRuntimeId,p.issuedAt,p.expiresAt,commandTuple(p.command)];}
|
||||
export function admissionTuple(a){return [a.schemaVersion,a.admissionId,packetTuple(a.packet),a.packetSha256,a.sourceObserverId,a.sourceObserverRuntimeId,a.provider,a.bifrostAdmissionReceiptId,a.authority,a.providerIdentityId];}
|
||||
// rmp-serde's compact serializer encodes the nested Rust wire-result struct as
|
||||
// this positional array inside the signed tuple. The separately hashed result
|
||||
// payload is encoded with named fields.
|
||||
export function wireResultTuple(v){return [v.schemaVersion,v.resultId,v.commandId,v.packetSha256,v.targetRuntimeId,v.disposition,v.consequenceKind,v.consequenceRef,v.completedAt,v.privateStateExposed,v.operatorStatus,v.stateStatus,v.coordinatorAction,v.brakeStatus,v.detail];}
|
||||
export function resultReceiptTuple(v){return [v.schemaVersion,wireResultTuple(v.result),v.resultPayloadSha256,v.commandId,v.packetSha256,v.targetRuntimeId,v.completedAt,v.providerIdentityId,v.privateStateExposed];}
|
||||
function parsePacket(v){v=exactRecord(v,["commandId","nonce","sourceEventId","sourceActorId","discordGuildId","discordChannelId","discordMessageId","targetRuntimeId","issuedAt","expiresAt","command"],"operator packet");for(const k of ["commandId","nonce","sourceEventId","sourceActorId","discordGuildId","discordChannelId","discordMessageId","targetRuntimeId"])text(v[k],k);date(v.issuedAt,"issuedAt");date(v.expiresAt,"expiresAt");parseCommand(v.command);return v;}
|
||||
function parseCommand(v){if(!v||typeof v!=="object"||Array.isArray(v))throw new Error("command must be an object.");if(v.kind==="status"||v.kind==="wake")return exactRecord(v,["kind"],"command");if(v.kind==="sleep"){exactRecord(v,["kind","reason"],"command");text(v.reason,"reason");if(v.reason.length>512)throw new Error("sleep reason exceeds 512 characters.");return v;}if(v.kind==="directive"){exactRecord(v,["kind","objective"],"command");text(v.objective,"objective");if(v.objective.length>4096)throw new Error("directive objective exceeds 4096 characters.");return v;}throw new Error("command kind is invalid.");}
|
||||
function commandTuple(v){if(v.kind==="sleep")return {kind:"sleep",reason:v.reason};if(v.kind==="directive")return {kind:"directive",objective:v.objective};return {kind:v.kind};}
|
||||
// rmp-serde's compact enum form is the snake-case unit variant string, or a
|
||||
// two-element [variant, sole-field] tuple for these one-field struct variants.
|
||||
function commandTuple(v){if(v.kind==="sleep")return ["sleep",v.reason];if(v.kind==="directive")return ["directive",v.objective];return v.kind;}
|
||||
function exactRecord(v,keys,label){if(!v||typeof v!=="object"||Array.isArray(v))throw new Error(`${label} must be an object.`);const actual=Object.keys(v).sort(),expected=[...keys].sort();if(actual.join("\0")!==expected.join("\0"))throw new Error(`${label} has unknown or missing fields.`);return v;}
|
||||
function text(v,label){if(typeof v!=="string"||!v.trim())throw new Error(`${label} is required.`);return v;}
|
||||
function exact(v,e,label){if(v!==e)throw new Error(`${label} must be ${e}.`);}
|
||||
function date(v,label){text(v,label);const n=Date.parse(v);if(!Number.isFinite(n)||!/T/.test(v))throw new Error(`${label} must be RFC3339.`);return n;}
|
||||
function date(v,label){text(v,label);if(!/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$/.test(v))throw new Error(`${label} must be RFC3339.`);const n=Date.parse(v);if(!Number.isFinite(n))throw new Error(`${label} must be RFC3339.`);return n;}
|
||||
function bytes(v,label,length){const normalized=v instanceof Uint8Array?v:Array.isArray(v)&&v.every(x=>Number.isInteger(x)&&x>=0&&x<=255)?Uint8Array.from(v):null;if(!normalized||normalized.length!==length)throw new Error(`${label} must be ${length} bytes.`);return normalized;}
|
||||
|
|
|
|||
|
|
@ -1,34 +1,59 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { resolve } from "node:path";
|
||||
import { admissionTuple, operatorAdmissionSchema, parseOperatorAdmission, parseOperatorRequest, parseOperatorResult, packetTuple } from "./epiphany-operator-command-documents.mjs";
|
||||
import { enrollSignedIdentity, loadSignedIdentity, signPurpose, trustAnchor } from "./bifrost-signed-identity.mjs";
|
||||
import { admissionTuple,operatorAdmissionSchema,parseOperatorAdmission,parseOperatorRequest,parseOperatorResult,parseOperatorResultReceipt,packetTuple } from "./epiphany-operator-command-documents.mjs";
|
||||
import { enrollSignedIdentity,loadSignedIdentity,signPurpose,trustAnchor } from "./bifrost-signed-identity.mjs";
|
||||
|
||||
export const OPERATOR_SIGNING_PURPOSE="bifrost.operator-command.delivery.v0";
|
||||
export const defaultOperatorIdentityPath=()=>process.env.BIFROST_EPIPHANY_OPERATOR_PRIVATE_KEY??(process.platform==="win32"?resolve(".bifrost","private","epiphany-operator-ed25519.seed"):"/var/lib/gamecult/bifrost/epiphany-operator/operator-ed25519.seed");
|
||||
|
||||
// The transport is deliberately a narrow port. The live CultNet RUDP request/reply
|
||||
// adapter is not yet present in Bifrost; callers must inject an authenticated port.
|
||||
export async function admitOperatorRequest({request,ownerDiscordId,actorLink,identity,encode,transport,expectedEpiphanyIdentityId,now=new Date(),ledger=new ImmutableOperatorLedger()}){
|
||||
parseOperatorRequest(request);
|
||||
if(request.sourceActorDiscordId!==ownerDiscordId)throw new Error("Epiphany operator command requires the exact configured Discord owner.");
|
||||
if(!actorLink||actorLink.guildId!==request.discordGuildId||actorLink.discordAuthorId!==request.sourceActorDiscordId||!actorLink.bifrostActorId?.trim()||!actorLink.heimdallActorRef?.trim())throw new Error("Epiphany operator command requires the exact existing Bifrost actor link and Heimdall actor reference.");
|
||||
const at=now instanceof Date?now:new Date(now);if(!Number.isFinite(at.getTime())||Date.parse(request.issuedAt)>at.getTime()||Date.parse(request.expiresAt)<at.getTime())throw new Error("Epiphany operator request is not currently valid.");
|
||||
const packet={commandId:request.commandId,nonce:request.nonce,sourceEventId:request.sourceEventId,sourceActorId:actorLink.bifrostActorId,discordGuildId:request.discordGuildId,discordChannelId:request.discordChannelId,discordMessageId:request.discordMessageId,targetRuntimeId:request.targetRuntimeId,issuedAt:request.issuedAt,expiresAt:request.expiresAt,command:request.command};
|
||||
const packetSha256=`sha256-${createHash("sha256").update(encode(packetTuple(packet))).digest("hex")}`;
|
||||
const admission={schemaName:"bifrost.operator_command.delivery",schemaVersion:operatorAdmissionSchema,admissionId:request.requestId,packet,packetSha256,sourceObserverId:"voidbot",sourceObserverRuntimeId:request.producerRuntimeId,provider:"bifrost",bifrostAdmissionReceiptId:request.requestId,authority:"exact_operator_command_only",providerIdentityId:identity.identityId,providerSignature:new Uint8Array()};
|
||||
admission.providerSignature=new Uint8Array(signPurpose(identity,OPERATOR_SIGNING_PURPOSE,encode(admissionTuple(admission))));parseOperatorAdmission(admission);
|
||||
ledger.admit(request.requestId,request,admission);
|
||||
if(!transport||typeof transport.execute!=="function")throw new Error("No Epiphany operator transport port is configured; the live CultNet RUDP adapter is missing.");
|
||||
const envelope=await transport.execute(admission);
|
||||
if(!envelope||envelope.providerIdentityId!==expectedEpiphanyIdentityId)throw new Error("Operator result did not cross the configured authenticated Epiphany transport boundary.");
|
||||
const result=parseOperatorResult(envelope.result);
|
||||
if(result.commandId!==packet.commandId||result.packetSha256!==packetSha256||result.targetRuntimeId!==packet.targetRuntimeId)throw new Error("Epiphany operator result is not bound to the admitted command.");
|
||||
ledger.complete(request.requestId,result,envelope.providerIdentityId);
|
||||
return {admission,result,resultProviderIdentityId:envelope.providerIdentityId};
|
||||
export class OperatorCommandPermanentRefusal extends Error { constructor(code,message){super(message);this.name="OperatorCommandPermanentRefusal";this.code=code;this.retryable=false;} }
|
||||
|
||||
export async function admitOperatorRequest({request,ownerDiscordId,actorLink,identity,encode,transport,now=new Date(),ledger}){
|
||||
requireDurableLedger(ledger);
|
||||
try{parseOperatorRequest(request);}catch(error){throw new OperatorCommandPermanentRefusal("invalid-request",error instanceof Error?error.message:String(error));}
|
||||
const existingRequest=await ledger.getRequest(request.requestId),existingAdmission=await ledger.getAdmission(request.requestId),existingCompletion=await ledger.getCompletion(request.commandId);
|
||||
if(Boolean(existingRequest)!==Boolean(existingAdmission))throw new Error("operator durable request/admission pair is incomplete");
|
||||
requireAuthenticatedTransport(transport);
|
||||
if(existingAdmission){
|
||||
await compareAdmission(ledger,request,existingAdmission);
|
||||
if(existingCompletion){
|
||||
parseOperatorResultReceipt(existingCompletion);
|
||||
if(existingCompletion.commandId!==existingAdmission.packet.commandId||existingCompletion.packetSha256!==existingAdmission.packetSha256||existingCompletion.targetRuntimeId!==existingAdmission.packet.targetRuntimeId)refuse("command-id-collision","operator completion is not bound to this exact durable admission");
|
||||
transport.verifyReceipt(existingAdmission,existingCompletion);
|
||||
return {admission:existingAdmission,result:existingCompletion.result,receipt:existingCompletion,resultProviderIdentityId:existingCompletion.providerIdentityId,recovered:true};
|
||||
}
|
||||
return executeAdmission({request,admission:existingAdmission,transport,ledger,recovered:true});
|
||||
}
|
||||
if(existingCompletion)refuse("command-id-collision","operator command id already belongs to a different durable admission");
|
||||
let admission;
|
||||
try{
|
||||
if(request.sourceActorDiscordId!==ownerDiscordId)refuse("foreign-owner","Epiphany operator command requires the exact configured Discord owner.");
|
||||
// This persisted Bifrost actor-link is the admitted mapping authority. Its
|
||||
// Heimdall reference is provenance only until a signed live claim exists.
|
||||
if(!actorLink||actorLink.guildId!==request.discordGuildId||actorLink.discordAuthorId!==request.sourceActorDiscordId||!actorLink.bifrostActorId?.trim())refuse("missing-actor-link","Epiphany operator command requires the exact admitted Bifrost actor link.");
|
||||
const at=now instanceof Date?now:new Date(now);if(!Number.isFinite(at.getTime())||Date.parse(request.issuedAt)>at.getTime()||Date.parse(request.expiresAt)<at.getTime())refuse("expired-new-request","A new Epiphany operator request is not currently valid.");
|
||||
const packet={commandId:request.commandId,nonce:request.nonce,sourceEventId:request.sourceEventId,sourceActorId:actorLink.bifrostActorId,discordGuildId:request.discordGuildId,discordChannelId:request.discordChannelId,discordMessageId:request.discordMessageId,targetRuntimeId:request.targetRuntimeId,issuedAt:request.issuedAt,expiresAt:request.expiresAt,command:request.command};
|
||||
const packetSha256=`sha256-${createHash("sha256").update(encode(packetTuple(packet))).digest("hex")}`;
|
||||
admission={schemaName:"bifrost.operator_command.delivery",schemaVersion:operatorAdmissionSchema,admissionId:request.requestId,packet,packetSha256,sourceObserverId:"voidbot",sourceObserverRuntimeId:request.producerRuntimeId,provider:"bifrost",bifrostAdmissionReceiptId:request.requestId,authority:"exact_operator_command_only",providerIdentityId:identity.identityId,providerSignature:new Uint8Array()};
|
||||
admission.providerSignature=new Uint8Array(signPurpose(identity,OPERATOR_SIGNING_PURPOSE,encode(admissionTuple(admission))));parseOperatorAdmission(admission);
|
||||
}catch(error){if(error instanceof OperatorCommandPermanentRefusal)throw error;throw new OperatorCommandPermanentRefusal("invalid-request",error instanceof Error?error.message:String(error));}
|
||||
await ledger.admit(request.requestId,request,admission);
|
||||
return executeAdmission({request,admission,transport,ledger,recovered:false});
|
||||
}
|
||||
|
||||
export class ImmutableOperatorLedger{#rows=new Map();admit(id,request,admission){const value={request,admission};const old=this.#rows.get(id);if(old&&stable(old)!==stable(value))throw new Error("operator request identity collision");if(!old)this.#rows.set(id,value);}complete(id,result,providerIdentityId){const row=this.#rows.get(id);if(!row)throw new Error("operator result has no admitted request");const next={...row,result,providerIdentityId};if(row.result&&stable(row)!==stable(next))throw new Error("operator result identity collision");this.#rows.set(id,next);}get(id){return this.#rows.get(id);}}
|
||||
async function executeAdmission({request,admission,transport,ledger,recovered}){
|
||||
const envelope=await transport.execute(admission);
|
||||
if(!envelope||envelope.providerIdentityId!==transport.trustedIdentityId)throw new Error("Operator result did not cross the configured authenticated Epiphany transport boundary.");
|
||||
const result=parseOperatorResult(envelope.result);
|
||||
if(result.commandId!==admission.packet.commandId||result.packetSha256!==admission.packetSha256||result.targetRuntimeId!==admission.packet.targetRuntimeId)throw new Error("Epiphany operator result is not bound to the admitted command.");
|
||||
await ledger.complete(request.requestId,result,envelope.providerIdentityId,envelope.receipt);
|
||||
return {admission,result,receipt:envelope.receipt,resultProviderIdentityId:envelope.providerIdentityId,recovered};
|
||||
}
|
||||
|
||||
async function compareAdmission(ledger,request,admission){try{await ledger.admit(request.requestId,request,admission);}catch(error){if(/collision/i.test(error instanceof Error?error.message:String(error)))refuse("request-identity-collision",error.message);throw error;}}
|
||||
function requireDurableLedger(ledger){for(const member of ["admit","complete","getRequest","getAdmission","getCompletion","terminalize","getTerminal"])if(!ledger||typeof ledger[member]!=="function")throw new Error("Epiphany operator admission requires the durable typed collision-ledger port.");}
|
||||
function requireAuthenticatedTransport(transport){if(!transport||typeof transport.execute!=="function"||typeof transport.verifyReceipt!=="function"||typeof transport.trustedIdentityId!=="string"||!transport.trustedIdentityId)throw new Error("No authenticated Epiphany operator CultNet RUDP transport is configured.");}
|
||||
function refuse(code,message){throw new OperatorCommandPermanentRefusal(code,message);}
|
||||
export async function enrollOperatorIdentity(path=defaultOperatorIdentityPath()){return enrollSignedIdentity(path,"Bifrost Epiphany operator identity");}
|
||||
export async function loadOperatorIdentity(path=defaultOperatorIdentityPath()){return loadSignedIdentity(path,"Bifrost Epiphany operator identity");}
|
||||
export {trustAnchor};
|
||||
function stable(v){return JSON.stringify(v,(_k,x)=>x instanceof Uint8Array?Buffer.from(x).toString("hex"):x);}
|
||||
|
|
|
|||
86
tools/epiphany-operator-cultnet-rudp.mjs
Normal file
86
tools/epiphany-operator-cultnet-rudp.mjs
Normal file
|
|
@ -0,0 +1,86 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { createSocket } from "node:dgram";
|
||||
import { createRequire } from "node:module";
|
||||
import { existsSync } from "node:fs";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { resolve } from "node:path";
|
||||
import {
|
||||
operatorAdmissionType,
|
||||
operatorResultReceiptType,
|
||||
parseOperatorAdmission,
|
||||
parseOperatorResultReceipt,
|
||||
resultReceiptTuple,
|
||||
} from "./epiphany-operator-command-documents.mjs";
|
||||
import { verifyPurpose } from "./bifrost-signed-identity.mjs";
|
||||
|
||||
export const EPIPHANY_RESULT_SIGNING_PURPOSE="epiphany.operator-command.sealed-result.v0";
|
||||
const RUDP_CONNECTION_ID=0xe91f0001;
|
||||
|
||||
export class EpiphanyOperatorCultNetRudpTransport {
|
||||
constructor({endpoint,runtimeId="bifrost-yggdrasil",trustedEpiphanyIdentity,timeoutMs=5000,cultnetRuntime,msgpackRuntime,socketFactory=createSocket,connectionFactory}={}){
|
||||
this.endpoint=parseEndpoint(endpoint);
|
||||
this.runtimeId=required(runtimeId,"Bifrost transport runtime id");
|
||||
this.trustedIdentity=parseTrustAnchor(trustedEpiphanyIdentity);
|
||||
if(!Number.isInteger(timeoutMs)||timeoutMs<100||timeoutMs>60000)throw new Error("Epiphany operator result timeout must be 100..=60000 milliseconds.");
|
||||
this.timeoutMs=timeoutMs;
|
||||
const loaded=(!cultnetRuntime||!msgpackRuntime)?loadCultNetRuntime():{};
|
||||
this.cultnet=cultnetRuntime??loaded.cultnet;
|
||||
this.msgpack=msgpackRuntime??loaded.msgpack;
|
||||
this.socketFactory=socketFactory;
|
||||
this.connectionFactory=connectionFactory??(options=>new this.cultnet.CultNetRudpSocketTransportConnection(options));
|
||||
this.tail=Promise.resolve();
|
||||
}
|
||||
|
||||
get trustedIdentityId(){return this.trustedIdentity.identityId;}
|
||||
verifyReceipt(admission,receipt){return verifySealedReceipt({receipt,admission,trustedIdentity:this.trustedIdentity,encode:this.msgpack.encode});}
|
||||
|
||||
execute(admission){
|
||||
const work=this.tail.then(()=>this.#executeOne(admission));
|
||||
this.tail=work.catch(()=>undefined);
|
||||
return work;
|
||||
}
|
||||
|
||||
async #executeOne(admission){
|
||||
parseOperatorAdmission(admission);
|
||||
const socket=this.socketFactory("udp4");
|
||||
await bindSocket(socket);
|
||||
const connection=this.connectionFactory({runtimeId:this.runtimeId,socket,mode:"client",remoteHost:this.endpoint.host,remotePort:this.endpoint.port,connectionId:RUDP_CONNECTION_ID,resendDelayMs:50,resendPollMs:10});
|
||||
try{
|
||||
connection.connect();
|
||||
await waitForConnected(connection,this.timeoutMs);
|
||||
const response=waitForReceipt(connection,this.msgpack,this.timeoutMs,admission);
|
||||
const message={schemaVersion:"cultnet.document_put_raw.v0",messageId:`operator-request-${admission.admissionId}`,document:{schemaId:operatorAdmissionType,recordKey:admission.admissionId,storedAt:admission.packet.issuedAt,payloadEncoding:"messagepack",payload:this.msgpack.encode(admission),sourceRuntimeId:admission.sourceObserverRuntimeId,sourceAgentId:admission.providerIdentityId,sourceRole:"bifrost-operator-admission",tags:["cultnet.transport.rudp.v0"]}};
|
||||
connection.send("schema",this.msgpack.encode(this.cultnet.encodeCultNetMessageForWire(message,"cultnet.schema.v0")));
|
||||
const receipt=await response;
|
||||
this.verifyReceipt(admission,receipt);
|
||||
return {providerIdentityId:receipt.providerIdentityId,result:receipt.result,receipt};
|
||||
}finally{connection.close();}
|
||||
}
|
||||
}
|
||||
|
||||
export async function loadEpiphanyOperatorTrustAnchor(path,{decode}={}){const runtime=decode?{decode}:loadCultNetRuntime().msgpack,value=runtime.decode(await readFile(path));if(!Array.isArray(value)||value.length!==6)throw new Error("Epiphany operator trust anchor must be the raw compact 6-field MessagePack crossing artifact.");return parseTrustAnchor({schemaVersion:value[0],identityId:value[1],publicKey:value[2],assurance:value[3],identityCreatedAt:value[4],sourceIdentityRecordSha256:value[5]});}
|
||||
|
||||
export function verifySealedReceipt({receipt,admission,trustedIdentity,encode}){
|
||||
parseOperatorAdmission(admission);parseOperatorResultReceipt(receipt);const anchor=parseTrustAnchor(trustedIdentity);
|
||||
if(receipt.providerIdentityId!==anchor.identityId)throw new Error("Epiphany sealed result used an untrusted host identity.");
|
||||
if(receipt.commandId!==admission.packet.commandId||receipt.packetSha256!==admission.packetSha256||receipt.targetRuntimeId!==admission.packet.targetRuntimeId)throw new Error("Epiphany sealed result receipt is not bound to the admitted command.");
|
||||
if(receipt.result.commandId!==receipt.commandId||receipt.result.packetSha256!==receipt.packetSha256||receipt.result.targetRuntimeId!==receipt.targetRuntimeId||receipt.result.completedAt!==receipt.completedAt)throw new Error("Epiphany sealed result receipt substituted its embedded application result.");
|
||||
// Expiry gates first admission, not exact crash recovery. Epiphany may seal
|
||||
// an already-admitted command after its lease while recovering the one
|
||||
// idempotent consequence; Bifrost's durable ledger owns whether such a
|
||||
// post-expiry retransmission was authorized.
|
||||
const completed=Date.parse(receipt.completedAt),issued=Date.parse(admission.packet.issuedAt);
|
||||
if(completed<issued)throw new Error("Epiphany sealed result predates the admitted command.");
|
||||
const digest=`sha256-${createHash("sha256").update(encode(receipt.result)).digest("hex")}`;
|
||||
if(digest!==receipt.resultPayloadSha256)throw new Error("Epiphany sealed result payload digest is invalid.");
|
||||
if(!verifyPurpose(anchor.publicKey,EPIPHANY_RESULT_SIGNING_PURPOSE,encode(resultReceiptTuple(receipt)),receipt.executorSignature))throw new Error("Epiphany sealed result signature is invalid.");
|
||||
return receipt;
|
||||
}
|
||||
|
||||
function waitForReceipt(connection,msgpack,timeoutMs,admission){return new Promise((resolveReceipt,reject)=>{const finish=(fn,value)=>{clearTimeout(timer);connection.off?.("frame",onFrame);connection.off?.("error",onError);fn(value);};const onError=error=>finish(reject,error);const onFrame=frame=>{try{if(frame.channelId!=="schema")throw new Error("Epiphany operator response arrived on a foreign CultNet channel.");const message=msgpack.decode(frame.payload);if(!message||message.schemaVersion!=="cultnet.document_put_raw.v0")throw new Error("Epiphany operator response is not a CultNet raw typed document.");const d=message.document;if(!d||d.schemaId!==operatorResultReceiptType||d.recordKey!==admission.packet.commandId||d.payloadEncoding!=="messagepack"||d.sourceRole!=="epiphany-operator-command-executor"||d.sourceRuntimeId!==admission.packet.targetRuntimeId)throw new Error("Epiphany operator response envelope substituted result authority.");const receipt=msgpack.decode(d.payload);if(d.sourceAgentId!==receipt?.providerIdentityId)throw new Error("Epiphany operator response envelope substituted executor identity.");finish(resolveReceipt,receipt);}catch(error){finish(reject,error);}};const timer=setTimeout(()=>finish(reject,new Error("Timed out waiting for the correlated Epiphany application result.")),timeoutMs);connection.on("frame",onFrame);connection.on("error",onError);});}
|
||||
async function waitForConnected(connection,timeoutMs){const started=Date.now();while(!connection.connected){if(Date.now()-started>=timeoutMs)throw new Error("Timed out establishing the Epiphany CultNet RUDP session.");await new Promise(done=>setTimeout(done,5));}}
|
||||
function bindSocket(socket){return new Promise((resolveBind,reject)=>{const fail=error=>{socket.off("listening",ready);reject(error)};const ready=()=>{socket.off("error",fail);resolveBind()};socket.once("error",fail);socket.once("listening",ready);socket.bind(0,"0.0.0.0");});}
|
||||
function parseEndpoint(value){required(value,"Epiphany operator RUDP endpoint");const url=new URL(value);if(url.protocol!=="rudp:"||!url.hostname||!url.port||(url.pathname&&url.pathname!=="/"))throw new Error("Epiphany operator endpoint must be rudp://host:port.");const port=Number(url.port);if(!Number.isInteger(port)||port<1||port>65535)throw new Error("Epiphany operator endpoint port is invalid.");return {host:url.hostname,port};}
|
||||
function parseTrustAnchor(value){if(!value||typeof value!=="object"||Array.isArray(value))throw new Error("Trusted Epiphany host identity is required.");const keys=Object.keys(value).sort(),expected=["assurance","identityCreatedAt","identityId","publicKey","schemaVersion","sourceIdentityRecordSha256"].sort();if(keys.join("\0")!==expected.join("\0")||value.schemaVersion!=="epiphany.host_identity_trust_anchor.v0"||typeof value.identityId!=="string"||!value.assurance?.trim()||!/^sha256-[0-9a-f]{64}$/.test(value.sourceIdentityRecordSha256)||!Number.isFinite(Date.parse(value.identityCreatedAt)))throw new Error("Trusted Epiphany host identity violates its exact schema.");value.publicKey=value.publicKey instanceof Uint8Array?value.publicKey:Array.isArray(value.publicKey)&&value.publicKey.every(x=>Number.isInteger(x)&&x>=0&&x<=255)?Uint8Array.from(value.publicKey):null;if(!value.publicKey||value.publicKey.length!==32)throw new Error("Trusted Epiphany host identity public key must be 32 bytes.");const derived=createHash("sha256").update(Buffer.from("epiphany.host-incarnation.identity.v0\0")).update(value.publicKey).digest("hex");if(derived!==value.identityId)throw new Error("Trusted Epiphany host identity id does not match its public key.");return value;}
|
||||
function loadCultNetRuntime(){const root=resolve(import.meta.dirname,".."),projects=resolve(root,".."),cult=resolve(process.env.VOIDBOT_CULTLIB_ROOT||resolve(projects,"CultLib")),pkg=resolve(cult,"packages","cultnet-ts","package.json");if(!existsSync(pkg))throw new Error(`CultNet TypeScript runtime is unavailable at ${pkg}.`);const requireCult=createRequire(pkg);return {cultnet:requireCult("cultnet-ts"),msgpack:requireCult("@msgpack/msgpack")};}
|
||||
function required(value,label){if(typeof value!=="string"||!value.trim())throw new Error(`${label} is required.`);return value;}
|
||||
19
tools/epiphany-operator-discord-outbox.mjs
Normal file
19
tools/epiphany-operator-discord-outbox.mjs
Normal file
|
|
@ -0,0 +1,19 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { createRequire } from "node:module";
|
||||
import { existsSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
import { buildOperatorDefinitions,operatorRequestTuple } from "./epiphany-operator-command-documents.mjs";
|
||||
|
||||
export class EpiphanyOperatorDiscordOutbox {
|
||||
constructor(cache,definition,encode){this.cache=cache;this.definition=definition;this.encode=encode;}
|
||||
static async open(path,{cultcacheRuntime,encode}={}){const runtime=cultcacheRuntime??loadCultCacheRuntime(),definition=buildOperatorDefinitions(runtime.defineDocumentType).discordDelivery,cache=runtime.CultCache.builder().withDocumentType(definition).withGenericStore(new runtime.SingleFileMessagePackBackingStore(path)).build();if(typeof encode!=="function")throw new Error("operator Discord outbox requires the canonical MessagePack encoder");await cache.pullAllBackingStores();return new EpiphanyOperatorDiscordOutbox(cache,definition,encode);}
|
||||
get(requestId){return this.cache.get(this.definition,requestId);}
|
||||
isExactDelivery(request){const value=this.get(request.requestId);return Boolean(value&&value.requestPayloadSha256===this.requestDigest(request)&&value.commandId===request.commandId&&value.discordGuildId===request.discordGuildId&&value.discordChannelId===request.discordChannelId&&value.discordInteractionId===request.sourceEventId&&value.targetRuntimeId===request.targetRuntimeId);}
|
||||
async publishCompleted(request,receipt){const value=this.base(request,{status:"completed",disposition:receipt.result.disposition,failureCode:"",detail:receipt.result.detail,operatorStatus:receipt.result.operatorStatus,stateStatus:receipt.result.stateStatus,coordinatorAction:receipt.result.coordinatorAction,brakeStatus:receipt.result.brakeStatus,sealedResultPayloadSha256:receipt.resultPayloadSha256,executorSignatureSha256:`sha256-${createHash("sha256").update(receipt.executorSignature).digest("hex")}`,resultProviderIdentityId:receipt.providerIdentityId,recordedAt:receipt.completedAt});return this.put(value);}
|
||||
async publishRefused(request,terminal){const value=this.base(request,{status:"refused",disposition:"",failureCode:terminal.failureCode,detail:terminal.failureCode,operatorStatus:"",stateStatus:"",coordinatorAction:"",brakeStatus:"",sealedResultPayloadSha256:"",executorSignatureSha256:"",resultProviderIdentityId:"",recordedAt:terminal.recordedAt});return this.put(value);}
|
||||
requestDigest(request){return `sha256-${createHash("sha256").update(this.encode(operatorRequestTuple(request))).digest("hex")}`;}
|
||||
base(request,result){return {schemaVersion:"bifrost.discord.epiphany_operator_delivery.v0",deliveryId:request.requestId,requestId:request.requestId,requestPayloadSha256:this.requestDigest(request),commandId:request.commandId,discordGuildId:request.discordGuildId,discordChannelId:request.discordChannelId,discordInteractionId:request.sourceEventId,targetRuntimeId:request.targetRuntimeId,...result,privateStateExposed:false};}
|
||||
async put(value){const current=this.get(value.requestId);if(current&&stable(current)!==stable(value))throw new Error("operator Discord delivery identity collision");if(!current)await this.cache.put(this.definition,value.deliveryId,value);return current??value;}
|
||||
}
|
||||
function loadCultCacheRuntime(){const root=resolve(import.meta.dirname,".."),projects=resolve(root,".."),cult=resolve(process.env.VOIDBOT_CULTLIB_ROOT||resolve(projects,"CultLib")),pkg=resolve(cult,"packages","cultcache-ts","package.json");if(!existsSync(pkg))throw new Error(`CultCache TypeScript runtime is unavailable at ${pkg}.`);const requireCult=createRequire(pkg);return requireCult("cultcache-ts");}
|
||||
function stable(v){return JSON.stringify(v,(_k,x)=>x instanceof Uint8Array?Buffer.from(x).toString("hex"):x);}
|
||||
20
tools/epiphany-operator-ledger.mjs
Normal file
20
tools/epiphany-operator-ledger.mjs
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
import { createRequire } from "node:module";
|
||||
import { existsSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
import { buildOperatorDefinitions } from "./epiphany-operator-command-documents.mjs";
|
||||
|
||||
export class DurableOperatorLedger {
|
||||
constructor(cache,definitions){this.cache=cache;this.definitions=definitions;}
|
||||
static async open(path,{cultcacheRuntime}={}){const runtime=cultcacheRuntime??loadCultCacheRuntime(),definitions=buildOperatorDefinitions(runtime.defineDocumentType),cache=runtime.CultCache.builder().withDocumentType(definitions.request).withDocumentType(definitions.admission).withDocumentType(definitions.sealedResult).withDocumentType(definitions.terminal).withGenericStore(new runtime.SingleFileMessagePackBackingStore(path)).build();await cache.pullAllBackingStores();return new DurableOperatorLedger(cache,definitions);}
|
||||
async admit(id,request,admission){if(id!==request.requestId||id!==admission.admissionId||request.commandId!==admission.packet.commandId)throw new Error("operator admission ledger key is not bound to request and admission identity");const currentRequest=this.cache.get(this.definitions.request,id),currentAdmission=this.cache.get(this.definitions.admission,id);if(currentRequest&&stable(currentRequest)!==stable(request)||currentAdmission&&stable(currentAdmission)!==stable(admission))throw new Error("operator request identity collision");if(!currentRequest)await this.cache.put(this.definitions.request,id,request);if(!currentAdmission)await this.cache.put(this.definitions.admission,id,admission);}
|
||||
async complete(id,result,providerIdentityId,receipt){const admission=this.cache.get(this.definitions.admission,id),request=this.cache.get(this.definitions.request,id);if(!admission||!request)throw new Error("operator result has no durable admitted request");if(!receipt||receipt.commandId!==result.commandId||receipt.providerIdentityId!==providerIdentityId||stable(receipt.result)!==stable(result)||receipt.commandId!==admission.packet.commandId||receipt.packetSha256!==admission.packetSha256||receipt.targetRuntimeId!==admission.packet.targetRuntimeId||request.commandId!==receipt.commandId)throw new Error("operator completion is not bound to its exact durable admission");const current=this.cache.get(this.definitions.sealedResult,result.commandId);if(current&&stable(current)!==stable(receipt))throw new Error("operator result identity collision");if(!current)await this.cache.put(this.definitions.sealedResult,result.commandId,receipt);}
|
||||
async terminalize(value){const current=this.cache.get(this.definitions.terminal,value.requestId);if(current&&stable(current)!==stable(value))throw new Error("operator terminal disposition identity collision");if(!current)await this.cache.put(this.definitions.terminal,value.requestId,value);}
|
||||
getAdmission(id){return this.cache.get(this.definitions.admission,id);}
|
||||
getRequest(id){return this.cache.get(this.definitions.request,id);}
|
||||
getCompletion(commandId){return this.cache.get(this.definitions.sealedResult,commandId);}
|
||||
getTerminal(requestId){return this.cache.get(this.definitions.terminal,requestId);}
|
||||
isExactCompletion(requestId,commandId){const request=this.getRequest(requestId),admission=this.getAdmission(requestId),completion=this.getCompletion(commandId);return Boolean(request&&admission&&completion&&request.commandId===commandId&&admission.packet.commandId===commandId&&completion.commandId===commandId&&completion.packetSha256===admission.packetSha256&&completion.targetRuntimeId===admission.packet.targetRuntimeId);}
|
||||
}
|
||||
|
||||
function loadCultCacheRuntime(){const root=resolve(import.meta.dirname,".."),projects=resolve(root,".."),cult=resolve(process.env.VOIDBOT_CULTLIB_ROOT||resolve(projects,"CultLib")),pkg=resolve(cult,"packages","cultcache-ts","package.json");if(!existsSync(pkg))throw new Error(`CultCache TypeScript runtime is unavailable at ${pkg}.`);const requireCult=createRequire(pkg);return requireCult("cultcache-ts");}
|
||||
function stable(v){return JSON.stringify(v,(_k,x)=>x instanceof Uint8Array?Buffer.from(x).toString("hex"):x);}
|
||||
75
tools/epiphany-operator-worker.mjs
Normal file
75
tools/epiphany-operator-worker.mjs
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
#!/usr/bin/env node
|
||||
import { createRequire } from "node:module";
|
||||
import { existsSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
import { buildOperatorDefinitions } from "./epiphany-operator-command-documents.mjs";
|
||||
import { buildFeedbackDefinitions } from "./persona-feedback-documents.mjs";
|
||||
import { admitOperatorRequest,loadOperatorIdentity,OperatorCommandPermanentRefusal } from "./epiphany-operator-commands.mjs";
|
||||
import { EpiphanyOperatorCultNetRudpTransport,loadEpiphanyOperatorTrustAnchor } from "./epiphany-operator-cultnet-rudp.mjs";
|
||||
import { DurableOperatorLedger } from "./epiphany-operator-ledger.mjs";
|
||||
import { EpiphanyOperatorDiscordOutbox } from "./epiphany-operator-discord-outbox.mjs";
|
||||
|
||||
const {command,options}=parseArgs(process.argv.slice(2));
|
||||
if(command==="help"){
|
||||
process.stdout.write("epiphany-operator-worker <once|serve> --request-store <cc> --private-store <cc> --ledger-store <cc> --delivery-store <cc> --private-key <seed> --epiphany-trust-anchor <msgpack> --epiphany-rudp <rudp://host:port> --owner-discord-id <id> [--runtime-id <id>] [--interval-ms <n>] [--timeout-ms <n>]\n");
|
||||
process.exit(0);
|
||||
}
|
||||
await main().catch(error=>{process.stderr.write(`${error instanceof Error?error.message:String(error)}\n`);process.exitCode=1;});
|
||||
|
||||
async function main(){
|
||||
if(!["once","serve"].includes(command))throw new Error("operator worker command must be once or serve");
|
||||
const runtime=loadRuntime();
|
||||
const operatorDefs=buildOperatorDefinitions(runtime.defineDocumentType);
|
||||
const feedbackDefs=buildFeedbackDefinitions(runtime.defineDocumentType);
|
||||
const requestStore=requiredOption("request-store","BIFROST_EPIPHANY_OPERATOR_REQUEST_STORE");
|
||||
const privateStore=requiredOption("private-store","BIFROST_PERSONA_FEEDBACK_PRIVATE_STORE");
|
||||
const ledgerStore=requiredOption("ledger-store","BIFROST_EPIPHANY_OPERATOR_LEDGER_STORE");
|
||||
const deliveryStore=requiredOption("delivery-store","BIFROST_EPIPHANY_OPERATOR_DISCORD_DELIVERY_STORE");
|
||||
const ownerDiscordId=requiredOption("owner-discord-id","BIFROST_EPIPHANY_OPERATOR_OWNER_DISCORD_ID");
|
||||
assertDistinctStores(requestStore,privateStore,ledgerStore,deliveryStore);
|
||||
const identity=await loadOperatorIdentity(resolve(requiredOption("private-key","BIFROST_EPIPHANY_OPERATOR_PRIVATE_KEY")));
|
||||
const anchor=await loadEpiphanyOperatorTrustAnchor(resolve(requiredOption("epiphany-trust-anchor","BIFROST_EPIPHANY_EXECUTOR_TRUST_ANCHOR")),{decode:runtime.decode});
|
||||
const transport=new EpiphanyOperatorCultNetRudpTransport({endpoint:requiredOption("epiphany-rudp","BIFROST_EPIPHANY_OPERATOR_RUDP"),runtimeId:option("runtime-id","BIFROST_RUNTIME_ID")||"bifrost-yggdrasil",trustedEpiphanyIdentity:anchor,timeoutMs:integerOption("timeout-ms",5000,100,60000),cultnetRuntime:runtime.cultnet,msgpackRuntime:runtime.msgpack});
|
||||
const ledger=await DurableOperatorLedger.open(resolve(ledgerStore),{cultcacheRuntime:runtime.cultcache});
|
||||
const outbox=await EpiphanyOperatorDiscordOutbox.open(resolve(deliveryStore),{cultcacheRuntime:runtime.cultcache,encode:runtime.encode});
|
||||
const requests=runtime.CultCache.builder().withDocumentType(operatorDefs.request).withGenericStore(new runtime.SingleFileMessagePackBackingStore(resolve(requestStore))).build();
|
||||
const privateCache=runtime.CultCache.builder().withDocumentType(feedbackDefs.actorLink).withGenericStore(new runtime.SingleFileMessagePackBackingStore(resolve(privateStore))).build();
|
||||
const interval=integerOption("interval-ms",500,100,60000);
|
||||
do{
|
||||
const output=await processPending({requests,privateCache,operatorDefs,feedbackDefs,ledger,outbox,ownerDiscordId,identity,transport,encode:runtime.encode});
|
||||
process.stdout.write(`${JSON.stringify({schemaVersion:"bifrost.epiphany_operator_worker_pulse.v0",status:output.every(x=>["completed","refused"].includes(x.status))?"completed":"degraded",processed:output.length,results:output,privateStateExposed:false})}\n`);
|
||||
if(command==="once")return;
|
||||
await new Promise(done=>setTimeout(done,interval));
|
||||
}while(true);
|
||||
}
|
||||
|
||||
async function processPending({requests,privateCache,operatorDefs,feedbackDefs,ledger,outbox,ownerDiscordId,identity,transport,encode}){
|
||||
await requests.pullAllBackingStores();await privateCache.pullAllBackingStores();
|
||||
const pending=requests.getAll(operatorDefs.request).filter(request=>!outbox.isExactDelivery(request)).sort((a,b)=>a.issuedAt.localeCompare(b.issuedAt));
|
||||
const output=[];
|
||||
for(const request of pending){
|
||||
const terminal=ledger.getTerminal(request.requestId);
|
||||
if(terminal){await outbox.publishRefused(request,terminal);output.push(summary(request,"refused","",terminal.failureCode));continue;}
|
||||
try{
|
||||
const actorLink=privateCache.get(feedbackDefs.actorLink,`${request.discordGuildId}:${request.sourceActorDiscordId}`);
|
||||
const result=await admitOperatorRequest({request,ownerDiscordId,actorLink,identity,encode,transport,ledger});
|
||||
await outbox.publishCompleted(request,result.receipt);
|
||||
output.push(summary(request,"completed",result.result.disposition,result.result.detail));
|
||||
}catch(error){
|
||||
if(error instanceof OperatorCommandPermanentRefusal){
|
||||
const value={schemaVersion:"bifrost.operator_command.terminal.v0",requestId:request.requestId,commandId:request.commandId,status:"refused",failureCode:error.code,recordedAt:new Date().toISOString(),retryable:false,privateStateExposed:false};
|
||||
await ledger.terminalize(value);await outbox.publishRefused(request,value);
|
||||
}
|
||||
output.push(summary(request,error instanceof OperatorCommandPermanentRefusal?"refused":"retryable-failure","",error instanceof OperatorCommandPermanentRefusal?error.code:"transport-or-ledger-unavailable"));
|
||||
}
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
function summary(request,status,disposition,detail){return {requestId:request.requestId,commandId:request.commandId,status,disposition,detail};}
|
||||
function loadRuntime(){const root=resolve(import.meta.dirname,".."),projects=resolve(root,".."),cult=resolve(process.env.VOIDBOT_CULTLIB_ROOT||resolve(projects,"CultLib")),cachePkg=resolve(cult,"packages","cultcache-ts","package.json"),netPkg=resolve(cult,"packages","cultnet-ts","package.json");if(!existsSync(cachePkg)||!existsSync(netPkg))throw new Error("CultLib CultCache/CultNet TypeScript runtimes are unavailable");const cr=createRequire(cachePkg),nr=createRequire(netPkg),cultcache=cr("cultcache-ts"),msgpack=nr("@msgpack/msgpack");return {cultcache,cultnet:nr("cultnet-ts"),msgpack,encode:msgpack.encode,decode:msgpack.decode,CultCache:cultcache.CultCache,SingleFileMessagePackBackingStore:cultcache.SingleFileMessagePackBackingStore,defineDocumentType:cultcache.defineDocumentType};}
|
||||
function parseArgs(values){const command=values.shift()??"help",options={},allowed=new Set(["request-store","private-store","ledger-store","delivery-store","private-key","epiphany-trust-anchor","epiphany-rudp","owner-discord-id","runtime-id","interval-ms","timeout-ms"]);while(values.length){const flag=values.shift();if(!flag?.startsWith("--")||!values.length)throw new Error(`invalid operator worker option ${flag??""}`);const key=flag.slice(2);if(!allowed.has(key))throw new Error(`unknown operator worker option --${key}`);if(Object.hasOwn(options,key))throw new Error(`duplicate operator worker option --${key}`);options[key]=values.shift();}return {command,options};}
|
||||
function option(name,environment){return options[name]??process.env[environment];}
|
||||
function requiredOption(name,environment){const value=option(name,environment);if(typeof value!=="string"||!value.trim())throw new Error(`operator worker requires --${name} or ${environment}`);return value;}
|
||||
function integerOption(name,fallback,min,max){const raw=options[name],value=raw===undefined?fallback:Number(raw);if(!Number.isInteger(value)||value<min||value>max)throw new Error(`operator worker --${name} must be ${min}..=${max}`);return value;}
|
||||
function assertDistinctStores(...paths){const resolved=paths.map(path=>resolve(path).toLowerCase());if(new Set(resolved).size!==resolved.length)throw new Error("operator request, actor-link, immutable ledger, and Discord delivery stores must be physically distinct");}
|
||||
|
|
@ -168,6 +168,9 @@ const motionCommandCultMeshUri = `cultmesh://${locatedService}/commands/motion`;
|
|||
const epiphanyOperatorRequestSchemaId = "voidbot.discord.epiphany_operator_request.v0";
|
||||
const epiphanyOperatorAdmissionSchemaId = "bifrost.operator_command.delivery.v0";
|
||||
const epiphanyOperatorResultSchemaId = "epiphany.operator_command.result.v0";
|
||||
const epiphanyOperatorSealedResultSchemaId = "epiphany.operator_command.sealed_result.v0";
|
||||
const epiphanyOperatorTerminalSchemaId = "bifrost.operator_command.terminal.v0";
|
||||
const epiphanyOperatorDiscordDeliverySchemaId = "bifrost.discord.epiphany_operator_delivery.v0";
|
||||
|
||||
const advertisementDefinition = defineDocumentType({
|
||||
type: documentType,
|
||||
|
|
@ -334,7 +337,8 @@ function buildAdvertisement(options) {
|
|||
endpoint("motion-surface", motionSurfaceCultMeshUri, "gamecult.eve.surface.v1", ["cultmesh", "governance", "motion"]),
|
||||
endpoint("motion-commands", motionCommandCultMeshUri, "bifrost.motion_command.v0", ["command", "cultmesh", "governance", "motion"]),
|
||||
endpoint("epiphany-operator-requests", `${locatedService}/commands/epiphany-operator`, epiphanyOperatorRequestSchemaId, ["command", "discord", "operator", "cultmesh"]),
|
||||
endpoint("epiphany-operator-results", `${locatedService}/receipts/epiphany-operator`, epiphanyOperatorResultSchemaId, ["receipt", "epiphany", "operator", "cultmesh"]),
|
||||
endpoint("epiphany-operator-results", `${locatedService}/receipts/epiphany-operator`, epiphanyOperatorSealedResultSchemaId, ["receipt", "epiphany", "operator", "cultnet", "rudp"]),
|
||||
endpoint("epiphany-operator-discord-deliveries", `${locatedService}/deliveries/discord/epiphany-operator`, epiphanyOperatorDiscordDeliverySchemaId, ["delivery", "discord", "operator", "cultcache"]),
|
||||
],
|
||||
routes: [
|
||||
route("cultmesh-store", ".bifrost/provider-store.cc", "local-cultmesh", true),
|
||||
|
|
@ -377,6 +381,9 @@ function buildAdvertisement(options) {
|
|||
schema(epiphanyOperatorRequestSchemaId, "voidbot.discord.epiphany_operator_request", "strict owner-authored Discord operator request inbox"),
|
||||
schema(epiphanyOperatorAdmissionSchemaId, "bifrost.operator_command.delivery", "Bifrost-signed expiring Epiphany operator admission"),
|
||||
schema(epiphanyOperatorResultSchemaId, "epiphany.operator_command.result", "Epiphany-origin command result returned across the authenticated transport boundary"),
|
||||
schema(epiphanyOperatorSealedResultSchemaId, "epiphany.operator_command.sealed_result", "Epiphany host-signed receipt binding the exact application result to command, packet, runtime, and payload digest"),
|
||||
schema(epiphanyOperatorTerminalSchemaId, "bifrost.operator_command.terminal", "Bifrost-owned immutable terminal refusal for a non-retryable operator request"),
|
||||
schema(epiphanyOperatorDiscordDeliverySchemaId, "bifrost.discord.epiphany_operator_delivery", "Bifrost-owned immutable Discord result delivery bound to the original request and sealed Epiphany result or terminal refusal"),
|
||||
],
|
||||
witnesses: [
|
||||
witness(".bifrost/provider-store.cc", `${schemaId}; ${surfaceSchemaId}; ${interfaceBindingSchemaId}; ${discordPostCommandSchemaId}; ${discordPostReceiptSchemaId}`, "current", "typed provider advertisement, operator surface, Eve interface binding, and Bifrost Discord command request/receipt store"),
|
||||
|
|
@ -389,6 +396,8 @@ function buildAdvertisement(options) {
|
|||
witness(".bifrost/bridge-receipts.cc", crossingReceiptSchemaId, "current", "canonical governed public crossing receipt history"),
|
||||
witness(".bifrost/eve-surfaces.cc", "gamecult.eve.surface.v1", "planned-export", "future product Eve/CultUI surfaces not yet in the provider store"),
|
||||
witness(".bifrost/epiphany-operator-requests.cc", epiphanyOperatorRequestSchemaId, "configured", "dedicated VoidBot-writable Bifrost operator request inbox; never an Epiphany store"),
|
||||
witness(".bifrost/epiphany-operator-ledger.cc", `${epiphanyOperatorAdmissionSchemaId}; ${epiphanyOperatorSealedResultSchemaId}; ${epiphanyOperatorTerminalSchemaId}`, "configured", "Bifrost-owned immutable admission, host-sealed application-result, and terminal-refusal ledger"),
|
||||
witness(".bifrost/epiphany-operator-discord-deliveries.cc", epiphanyOperatorDiscordDeliverySchemaId, "configured", "Bifrost-writable and VoidBot-readable immutable operator result delivery outbox"),
|
||||
],
|
||||
surfaces: [
|
||||
surface("bifrost", "Bifrost Operator Dashboard", "gamecult.bifrost.surface.operator", "gamecult.eve.surface_state.v1", ".bifrost/provider-store.cc", "eve", [
|
||||
|
|
@ -500,7 +509,7 @@ function buildAdvertisement(options) {
|
|||
]),
|
||||
boundary("epiphany-operator", "Bifrost admission; Epiphany execution", [
|
||||
"admit exact owner-only status, sleep, wake, and directive requests",
|
||||
"require an existing Bifrost actor link and Heimdall actor reference",
|
||||
"require the exact configured Discord owner and persisted Bifrost actor-link mapping; Heimdall reference is provenance until signed claims are live",
|
||||
"sign Rust-compatible expiring operator admissions with a dedicated Ed25519 identity",
|
||||
"relay only Epiphany-origin results bound to the admitted command",
|
||||
], [
|
||||
|
|
|
|||
Loading…
Reference in a new issue