diff --git a/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts b/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts index 9dbd574..82ff173 100644 --- a/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts +++ b/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts @@ -1,11 +1,55 @@ -import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; +import { + ApiProblem, + apiSubject, + problemResponse, + requestTraceId, +} from "@/lib/api-context"; import { SyntheticCleanupDomainService } from "@/lib/synthetic-cleanup-domain"; +/** Maintenance manifests can be large; cap body size before buffering JSON. */ +const MAX_SYNTHETIC_CLEANUP_BODY_BYTES = 1_048_576; + +function assertSyntheticCleanupBodySize(request: Request) { + const contentLength = request.headers.get("content-length"); + if (contentLength !== null) { + const length = Number(contentLength); + if ( + !Number.isSafeInteger(length) || + length < 0 || + length > MAX_SYNTHETIC_CLEANUP_BODY_BYTES + ) { + throw new ApiProblem( + 413, + "Payload too large", + `Synthetic cleanup request body must be at most ${MAX_SYNTHETIC_CLEANUP_BODY_BYTES} bytes.`, + ); + } + } +} + export async function POST(request: Request) { const traceId = requestTraceId(request); try { const subject = await apiSubject(request); - const body = await request.json(); + assertSyntheticCleanupBodySize(request); + const raw = await request.text(); + if (Buffer.byteLength(raw, "utf8") > MAX_SYNTHETIC_CLEANUP_BODY_BYTES) { + throw new ApiProblem( + 413, + "Payload too large", + `Synthetic cleanup request body must be at most ${MAX_SYNTHETIC_CLEANUP_BODY_BYTES} bytes.`, + ); + } + let body: unknown; + try { + body = raw.length === 0 ? {} : JSON.parse(raw); + } catch { + throw new ApiProblem( + 400, + "Request failed", + "Request body must be valid JSON.", + ); + } const data = await new SyntheticCleanupDomainService().execute( subject, body, diff --git a/apps/web/lib/agent-direct-message-domain.ts b/apps/web/lib/agent-direct-message-domain.ts index 10b561d..10666cf 100644 --- a/apps/web/lib/agent-direct-message-domain.ts +++ b/apps/web/lib/agent-direct-message-domain.ts @@ -1,330 +1,4 @@ -import { createHash } from "node:crypto"; -import { requireCapability, type AuthorisationSubject } from "@muster/authz"; -import { redactObservationText } from "@muster/config"; -import { - appendAuditEvent, - database, - newId, - schema, - writeOutbox, -} from "@muster/database"; -import { and, eq, gt, isNull, or } from "drizzle-orm"; - -type Db = ReturnType; - -export type DirectMessageInvocation = - | { - handled: true; - queued: false; - duplicate: false; - reason: "agent_count" | "agent_unavailable"; - } - | { - handled: true; - queued: true; - duplicate: boolean; - agentId: string; - agentRunId: string; - status: string; - }; - -export class AgentDirectMessageDomainService { - constructor(private readonly db: Db = database()) {} - - async maybeQueue( - subject: AuthorisationSubject, - input: { messageId: string; roomId: string }, - traceId: string, - ): Promise { - const [source] = await this.db - .select({ - plainText: schema.messages.plainText, - relatedInvestigationId: schema.messages.relatedInvestigationId, - }) - .from(schema.messages) - .innerJoin( - schema.rooms, - and( - eq(schema.rooms.organisationId, subject.organisationId), - eq(schema.rooms.id, schema.messages.roomId), - eq(schema.rooms.roomType, "direct"), - isNull(schema.rooms.archivedAt), - ), - ) - .innerJoin( - schema.roomMemberships, - and( - eq(schema.roomMemberships.organisationId, subject.organisationId), - eq(schema.roomMemberships.roomId, schema.messages.roomId), - eq(schema.roomMemberships.actorId, subject.actorId), - or( - isNull(schema.roomMemberships.accessExpiresAt), - gt(schema.roomMemberships.accessExpiresAt, new Date()), - ), - ), - ) - .innerJoin( - schema.actors, - and( - eq(schema.actors.organisationId, subject.organisationId), - eq(schema.actors.id, schema.messages.authorActorId), - eq(schema.actors.id, subject.actorId), - eq(schema.actors.actorType, "human"), - eq(schema.actors.status, "active"), - ), - ) - .where( - and( - eq(schema.messages.organisationId, subject.organisationId), - eq(schema.messages.id, input.messageId), - eq(schema.messages.roomId, input.roomId), - eq(schema.messages.messageType, "text"), - isNull(schema.messages.deletedAt), - ), - ) - .limit(1); - if (!source) return null; - - const members = await this.db - .select({ - actorId: schema.actors.id, - actorStatus: schema.actors.status, - definitionId: schema.agentDefinitions.id, - definitionStatus: schema.agentDefinitions.status, - killSwitch: schema.agentDefinitions.killSwitch, - allowedRooms: schema.agentDefinitions.allowedRooms, - runtime: schema.agentDefinitions.runtime, - model: schema.agentDefinitions.model, - promptVersion: schema.agentDefinitions.systemPromptVersion, - maximumRuntimeSeconds: schema.agentDefinitions.maximumRuntimeSeconds, - maximumTokenBudget: schema.agentDefinitions.maximumTokenBudget, - maximumCostCents: schema.agentDefinitions.maximumCostCents, - }) - .from(schema.roomMemberships) - .innerJoin( - schema.actors, - and( - eq(schema.actors.organisationId, subject.organisationId), - eq(schema.actors.id, schema.roomMemberships.actorId), - eq(schema.actors.actorType, "agent"), - ), - ) - .leftJoin( - schema.agentDefinitions, - and( - eq(schema.agentDefinitions.organisationId, subject.organisationId), - eq(schema.agentDefinitions.id, schema.actors.id), - ), - ) - .where( - and( - eq(schema.roomMemberships.organisationId, subject.organisationId), - eq(schema.roomMemberships.roomId, input.roomId), - or( - isNull(schema.roomMemberships.accessExpiresAt), - gt(schema.roomMemberships.accessExpiresAt, new Date()), - ), - ), - ); - if (members.length === 0) return null; - - const active = members.filter( - (member) => - member.actorStatus === "active" && - member.definitionId !== null && - member.definitionStatus === "active" && - member.killSwitch === false, - ); - if (active.length !== 1) { - return { - handled: true, - queued: false, - duplicate: false, - reason: "agent_count", - }; - } - const agent = active[0]!; - if ( - !Array.isArray(agent.allowedRooms) || - !agent.allowedRooms.includes(input.roomId) || - !agent.definitionId || - !agent.runtime || - !agent.model || - !agent.promptVersion || - agent.maximumRuntimeSeconds === null || - agent.maximumTokenBudget === null || - agent.maximumCostCents === null - ) { - return { - handled: true, - queued: false, - duplicate: false, - reason: "agent_unavailable", - }; - } - - const agentId = agent.definitionId; - const runtime = agent.runtime; - const model = agent.model; - const promptVersion = agent.promptVersion; - const maximumRuntimeSeconds = agent.maximumRuntimeSeconds; - const maximumTokenBudget = agent.maximumTokenBudget; - const maximumCostCents = agent.maximumCostCents; - requireCapability(subject, "agents.invoke"); - const idempotencyKey = `agent-direct-message:${input.messageId}`; - const inputHash = createHash("sha256") - .update( - JSON.stringify({ - messageId: input.messageId, - roomId: input.roomId, - plainText: source.plainText, - }), - ) - .digest("hex"); - const deadlineAt = new Date(Date.now() + maximumRuntimeSeconds * 1_000); - - return this.db.transaction(async (tx) => { - const eligible = await tx - .select({ - id: schema.agentDefinitions.id, - allowedRooms: schema.agentDefinitions.allowedRooms, - }) - .from(schema.agentDefinitions) - .innerJoin( - schema.actors, - and( - eq(schema.actors.organisationId, subject.organisationId), - eq(schema.actors.id, schema.agentDefinitions.id), - eq(schema.actors.status, "active"), - ), - ) - .innerJoin( - schema.roomMemberships, - and( - eq(schema.roomMemberships.organisationId, subject.organisationId), - eq(schema.roomMemberships.roomId, input.roomId), - eq(schema.roomMemberships.actorId, schema.agentDefinitions.id), - or( - isNull(schema.roomMemberships.accessExpiresAt), - gt(schema.roomMemberships.accessExpiresAt, new Date()), - ), - ), - ) - .where( - and( - eq(schema.agentDefinitions.organisationId, subject.organisationId), - eq(schema.agentDefinitions.status, "active"), - eq(schema.agentDefinitions.killSwitch, false), - ), - ); - if ( - eligible.length !== 1 || - eligible[0]?.id !== agentId || - !Array.isArray(eligible[0].allowedRooms) || - !eligible[0].allowedRooms.includes(input.roomId) - ) { - throw new Error("Direct-message agent is unavailable"); - } - - const [inserted] = await tx - .insert(schema.agentRuns) - .values({ - id: newId(), - agentId, - organisationId: subject.organisationId, - roomId: input.roomId, - investigationId: source.relatedInvestigationId, - requestedByActorId: subject.actorId, - trigger: "direct_message", - status: "queued", - request: { - kind: "direct_message", - sourceMessageId: input.messageId, - humanRequest: source.plainText, - traceId, - }, - progress: { stage: "queued", percent: 0 }, - deadlineAt, - inputHash, - promptVersion, - runtime, - model, - maximumRuntimeSeconds, - maximumTokenBudget, - maximumCostCents, - idempotencyKey, - }) - .onConflictDoNothing() - .returning(); - const run = - inserted ?? - ( - await tx - .select() - .from(schema.agentRuns) - .where( - and( - eq(schema.agentRuns.organisationId, subject.organisationId), - eq(schema.agentRuns.idempotencyKey, idempotencyKey), - ), - ) - .limit(1) - )[0]; - if ( - !run || - run.agentId !== agentId || - run.roomId !== input.roomId || - run.requestedByActorId !== subject.actorId || - run.inputHash !== inputHash - ) { - throw new Error("Direct-message run idempotency conflict"); - } - if (inserted) { - await tx.insert(schema.agentRunEvents).values({ - id: newId(), - organisationId: subject.organisationId, - runId: run.id, - eventType: "queued", - message: "Direct-message agent run queued", - payload: { - trigger: "direct_message", - sourceMessageId: input.messageId, - roomId: input.roomId, - }, - }); - await appendAuditEvent(tx, { - organisationId: subject.organisationId, - actorId: subject.actorId, - actorType: "human", - action: "agent.run.queued", - targetType: "agent_run", - targetId: run.id, - metadata: { - trigger: "direct_message", - sourceMessageId: input.messageId, - roomId: input.roomId, - }, - traceId: redactObservationText(traceId), - }); - await writeOutbox(tx, { - organisationId: subject.organisationId, - eventType: "agent.run.queued", - aggregateType: "agent_run", - aggregateId: run.id, - queueName: "muster-agents", - payload: { runId: run.id }, - idempotencyKey: `agent.run.queued:${run.id}`, - traceId: redactObservationText(traceId), - }); - } - return { - handled: true, - queued: true, - duplicate: !inserted, - agentId: run.agentId, - agentRunId: run.id, - status: run.status, - }; - }); - } -} +export { + AgentDirectMessageDomainService, + type DirectMessageInvocation, +} from "@muster/rooms"; diff --git a/apps/web/lib/connector-domain.ts b/apps/web/lib/connector-domain.ts index 7ac6ecb..0ed49e9 100644 --- a/apps/web/lib/connector-domain.ts +++ b/apps/web/lib/connector-domain.ts @@ -94,6 +94,8 @@ export class ConnectorDomainService { status: "configured", mock: request.testMode, configuration: { ...publicConfiguration, authType: auth.type }, + // Reactivate archived connectors so they reappear in list(). + archivedAt: null, updatedAt: new Date(), }) .where( diff --git a/apps/worker/package.json b/apps/worker/package.json index 37a7fa9..9694ed9 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -13,12 +13,14 @@ }, "dependencies": { "@muster/agents": "workspace:*", + "@muster/authz": "workspace:*", "@muster/config": "workspace:*", "@muster/agent-harness": "workspace:*", "@muster/contracts": "workspace:*", "@muster/database": "workspace:*", "@muster/evidence": "workspace:*", "@muster/integrations": "workspace:*", + "@muster/rooms": "workspace:*", "bullmq": "5.81.2", "drizzle-orm": "0.45.2", "nodemailer": "9.0.3", diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index ecfea1b..2fb4858 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -60,6 +60,15 @@ import { import { appendResearchTerminalMessage } from "./research-status.ts"; import { queueDueParkerReports } from "./parker-scheduler.ts"; import { processParkerReport } from "./parker-report.ts"; +import { + AgentDirectMessageDomainService, + type DirectMessageInvocation, +} from "@muster/rooms"; +import { + capabilities, + type AuthorisationSubject, + type Capability, +} from "@muster/authz"; const redisUrl = new URL(process.env.REDIS_URL ?? "redis://localhost:6379"); const connection = { @@ -174,14 +183,28 @@ const authoritativeProcessor: Processor = async (job) => { } if ( job.queueName === "muster-agents" && - job.name !== "report.generate.queued" + job.name === "agent.direct_message.evaluate" + ) { + await processDirectMessageEvaluate( + job.data.organisationId, + job.data.aggregateId, + job.data.traceId, + ); + } + if ( + job.queueName === "muster-agents" && + job.name !== "report.generate.queued" && + job.name !== "agent.direct_message.evaluate" ) { const gatewayToken = process.env.MUSTER_AGENT_GATEWAY_TOKEN?.trim(); if (!gatewayToken) throw new Error("Agent gateway token is not configured"); const response = await fetch( `${process.env.AGENT_GATEWAY_URL ?? "http://agent-gateway:3002"}/v1/runs/dispatch`, { - headers: { authorization: `Bearer ${gatewayToken}` }, + headers: { + authorization: `Bearer ${gatewayToken}`, + "x-muster-organisation-id": job.data.organisationId, + }, method: "POST", signal: AbortSignal.timeout(10_000), }, @@ -2108,6 +2131,88 @@ for (const name of queueNames.filter((queue) => queue !== "muster-outbox")) { workers.push(worker); } + +async function processDirectMessageEvaluate( + organisationId: string, + messageId: string, + traceId: string, +) { + const db = database(); + const [message] = await db + .select({ + roomId: schema.messages.roomId, + authorActorId: schema.messages.authorActorId, + }) + .from(schema.messages) + .where( + and( + eq(schema.messages.organisationId, organisationId), + eq(schema.messages.id, messageId), + ), + ) + .limit(1); + if (!message) return; + + const [actor] = await db + .select({ + id: schema.actors.id, + organisationId: schema.actors.organisationId, + capabilityAssignments: schema.actors.capabilityAssignments, + status: schema.actors.status, + actorType: schema.actors.actorType, + }) + .from(schema.actors) + .where( + and( + eq(schema.actors.organisationId, organisationId), + eq(schema.actors.id, message.authorActorId), + eq(schema.actors.actorType, "human"), + eq(schema.actors.status, "active"), + ), + ) + .limit(1); + if (!actor) return; + + const assigned = Array.isArray(actor.capabilityAssignments) + ? actor.capabilityAssignments.filter( + (value): value is Capability => + typeof value === "string" && + capabilities.includes(value as Capability), + ) + : []; + const subject: AuthorisationSubject = { + actorId: actor.id, + organisationId: actor.organisationId, + capabilities: new Set(assigned), + }; + + let result: DirectMessageInvocation | null = null; + try { + result = await new AgentDirectMessageDomainService(db).maybeQueue( + subject, + { messageId, roomId: message.roomId }, + traceId, + ); + } catch (error) { + // Capability or eligibility failures are terminal for this redrive; do not + // poison the queue. Log and acknowledge the outbox job. + jsonLog("warn", "agent.direct_message.evaluate.skipped", { + organisationId, + messageId, + error: error instanceof Error ? error.message : "evaluate failed", + }); + return; + } + if (result?.queued) { + jsonLog("info", "agent.direct_message.evaluate.queued", { + organisationId, + messageId, + agentRunId: result.agentRunId, + duplicate: result.duplicate, + }); + } +} + async function dispatchOutbox() { const db = database(); const events = await claimOutboxBatch(db, 100); diff --git a/apps/worker/src/research-config.test.ts b/apps/worker/src/research-config.test.ts index eb31f12..b9af0cd 100644 --- a/apps/worker/src/research-config.test.ts +++ b/apps/worker/src/research-config.test.ts @@ -8,14 +8,29 @@ function repositoryFile(path: string) { return readFileSync(new URL(path, `file://${repositoryRoot}/`), "utf8"); } +function extractComposeService(compose: string, service: string) { + const match = compose.match( + new RegExp( + `\\n ${service}:\\n([\\s\\S]*?)(?=\\n [a-zA-Z][\\w-]*:|\\n[a-zA-Z]|$)`, + ), + ); + if (!match?.[1]) { + throw new Error(`Compose service block not found: ${service}`); + } + return match[1]; +} + describe("Alfie homelab research configuration", () => { it("passes an explicitly configured feed-origin allowlist to web and worker only", () => { const compose = repositoryFile("deploy/docker/docker-compose.homelab.yml"); - expect( - compose.match( - /MUSTER_RESEARCH_ALLOWED_FEED_ORIGINS: \$\{MUSTER_RESEARCH_ALLOWED_FEED_ORIGINS:-\}/g, - ), - ).toHaveLength(2); + const allowlist = + /MUSTER_RESEARCH_ALLOWED_FEED_ORIGINS: \$\{MUSTER_RESEARCH_ALLOWED_FEED_ORIGINS:-\}/; + const web = extractComposeService(compose, "web"); + const worker = extractComposeService(compose, "worker"); + const postgres = extractComposeService(compose, "postgres"); + expect(web).toMatch(allowlist); + expect(worker).toMatch(allowlist); + expect(postgres).not.toMatch(/MUSTER_RESEARCH_ALLOWED_FEED_ORIGINS/); }); it("documents an empty safe default without enabling a mock origin", () => { diff --git a/apps/worker/src/research-status.ts b/apps/worker/src/research-status.ts index d35f06c..d298815 100644 --- a/apps/worker/src/research-status.ts +++ b/apps/worker/src/research-status.ts @@ -1,5 +1,5 @@ import { database, newId, schema, writeOutbox } from "@muster/database"; -import { and, eq } from "drizzle-orm"; +import { and, eq, isNull } from "drizzle-orm"; type Transaction = Parameters< Parameters["transaction"]>[0] @@ -38,6 +38,14 @@ export async function appendResearchTerminalMessage( eq(schema.researchWatchlists.id, schema.researchRuns.watchlistId), ), ) + .innerJoin( + schema.rooms, + and( + eq(schema.rooms.organisationId, input.organisationId), + eq(schema.rooms.id, schema.researchWatchlists.roomId), + isNull(schema.rooms.archivedAt), + ), + ) .innerJoin( schema.agentRuns, and( @@ -46,6 +54,14 @@ export async function appendResearchTerminalMessage( eq(schema.agentRuns.status, expectedStatus), ), ) + .innerJoin( + schema.roomMemberships, + and( + eq(schema.roomMemberships.organisationId, input.organisationId), + eq(schema.roomMemberships.roomId, schema.researchWatchlists.roomId), + eq(schema.roomMemberships.actorId, schema.agentRuns.agentId), + ), + ) .where( and( eq(schema.researchRuns.organisationId, input.organisationId), diff --git a/docker-compose.yml b/docker-compose.yml index e78ee73..801bf90 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -29,7 +29,7 @@ x-muster-environment: &muster-environment MUSTER_MOCK_INTEGRATIONS: "true" MUSTER_AGENT_RUNTIME: ${MUSTER_AGENT_RUNTIME:-codex} AGENT_GATEWAY_URL: http://agent-gateway:3002 - MUSTER_AGENT_GATEWAY_TOKEN: ${MUSTER_AGENT_GATEWAY_TOKEN:-local-agent-gateway-token-change-me-32-characters} + MUSTER_AGENT_GATEWAY_TOKEN: ${MUSTER_AGENT_GATEWAY_TOKEN:?set MUSTER_AGENT_GATEWAY_TOKEN to a unique secret} CODEX_HOME: /var/lib/muster/codex services: diff --git a/packages/audit/src/audit.test.ts b/packages/audit/src/audit.test.ts index 8fd3a49..e6bfb56 100644 --- a/packages/audit/src/audit.test.ts +++ b/packages/audit/src/audit.test.ts @@ -153,4 +153,33 @@ describe("audit chain", () => { legacyApprovalIdOmissions: [], }); }); + + it("accepts legacy undefined approvalId on integration.action.failed", () => { + const historicalInput = { + ...base, + action: "integration.action.failed", + metadata: { + integrationId: "integration", + operation: "alerts.list", + capability: "alerts.read", + approvalId: undefined, + }, + }; + const historicalEvent = { + ...historicalInput, + metadata: { + integrationId: "integration", + operation: "alerts.list", + capability: "alerts.read", + }, + eventHash: hashAuditEvent(historicalInput), + }; + + expect(verifyAuditIntegrity([historicalEvent])).toMatchObject({ + outcome: "legacy-compatible-not-strict", + strict: { valid: false, brokenAt: 1 }, + legacyCompatible: { valid: true }, + legacyApprovalIdOmissions: [{ sequence: 1 }], + }); + }); }); diff --git a/packages/audit/src/index.ts b/packages/audit/src/index.ts index cf22795..d9f27a0 100644 --- a/packages/audit/src/index.ts +++ b/packages/audit/src/index.ts @@ -34,6 +34,7 @@ type PersistedAuditEvent = HashableAuditEvent & { eventHash: string }; const legacyApprovalIdActions = new Set([ "integration.action.queued", "integration.action.succeeded", + "integration.action.failed", ]); function canonical(value: unknown): string { diff --git a/packages/database/src/verify-audit-integrity.ts b/packages/database/src/verify-audit-integrity.ts index 932dc93..2fc60bd 100644 --- a/packages/database/src/verify-audit-integrity.ts +++ b/packages/database/src/verify-audit-integrity.ts @@ -1,12 +1,19 @@ import { asc, eq } from "drizzle-orm"; +import { z } from "zod"; import { verifyAuditIntegrity } from "@muster/audit"; import { closeDatabase, database, schema } from "./index.ts"; -const organisationId = process.env.MUSTER_AUDIT_ORGANISATION_ID; +const organisationIdResult = z + .string() + .uuid() + .safeParse(process.env.MUSTER_AUDIT_ORGANISATION_ID); -if (!organisationId) { - throw new Error("MUSTER_AUDIT_ORGANISATION_ID is required"); +if (!organisationIdResult.success) { + throw new Error( + "MUSTER_AUDIT_ORGANISATION_ID is required and must be a UUID", + ); } +const organisationId = organisationIdResult.data; try { const events = await database() diff --git a/packages/evidence/src/object-storage.test.ts b/packages/evidence/src/object-storage.test.ts index 53b0ebd..1cb5550 100644 --- a/packages/evidence/src/object-storage.test.ts +++ b/packages/evidence/src/object-storage.test.ts @@ -10,6 +10,8 @@ describe("versioned object storage", () => { it("downloads the exact immutable version", async () => { vi.stubEnv("OBJECT_STORAGE_ENDPOINT", "http://127.0.0.1:9000"); vi.stubEnv("OBJECT_STORAGE_BUCKET", "muster-evidence"); + vi.stubEnv("OBJECT_STORAGE_ACCESS_KEY", "test-access-key"); + vi.stubEnv("OBJECT_STORAGE_SECRET_KEY", "test-secret-key"); const body = new TextEncoder().encode("exact version"); const fetchMock = vi.fn().mockResolvedValue( new Response(body, { diff --git a/packages/evidence/src/object-storage.ts b/packages/evidence/src/object-storage.ts index 1d1f152..185397d 100644 --- a/packages/evidence/src/object-storage.ts +++ b/packages/evidence/src/object-storage.ts @@ -29,6 +29,8 @@ export interface CleanupObjectStorage extends ContentObjectStorage { deleteObject(storageKey: string, versionId: string): Promise; } +const DEFAULT_OBJECT_STORAGE_TIMEOUT_MS = 30_000; + function sha256(value: string | Uint8Array) { return createHash("sha256").update(value).digest("hex"); } @@ -115,16 +117,29 @@ function signingHeaders( }; } +function requireEnv(name: string) { + const value = process.env[name]?.trim(); + if (!value) { + throw new Error(`${name} is required for object storage`); + } + return value; +} + function storageConfiguration() { return { endpoint: process.env.OBJECT_STORAGE_ENDPOINT ?? "http://127.0.0.1:9000", bucket: process.env.OBJECT_STORAGE_BUCKET ?? "muster-evidence", region: process.env.OBJECT_STORAGE_REGION ?? "us-east-1", - accessKey: process.env.OBJECT_STORAGE_ACCESS_KEY ?? "muster", - secretKey: process.env.OBJECT_STORAGE_SECRET_KEY ?? "local-minio-secret", + accessKey: requireEnv("OBJECT_STORAGE_ACCESS_KEY"), + secretKey: requireEnv("OBJECT_STORAGE_SECRET_KEY"), }; } +function requestSignal(signal?: AbortSignal) { + if (signal) return signal; + return AbortSignal.timeout(DEFAULT_OBJECT_STORAGE_TIMEOUT_MS); +} + export async function checkObjectStorage(signal?: AbortSignal) { const { endpoint, bucket, region, accessKey, secretKey } = storageConfiguration(); @@ -140,7 +155,7 @@ export async function checkObjectStorage(signal?: AbortSignal) { accessKey, secretKey, ), - ...(signal ? { signal } : {}), + signal: requestSignal(signal), }); if (!response.ok) { throw new Error( @@ -166,6 +181,7 @@ export const defaultObjectStorage: CleanupObjectStorage = { object.contentType, ), body: Buffer.from(object.body), + signal: requestSignal(), }); if (!response.ok) { throw new Error( @@ -188,6 +204,7 @@ export const defaultObjectStorage: CleanupObjectStorage = { accessKey, secretKey, ), + signal: requestSignal(), }); if (!response.ok) { throw new Error( @@ -214,6 +231,7 @@ export const defaultObjectStorage: CleanupObjectStorage = { accessKey, secretKey, ), + signal: requestSignal(), }); if (response.status === 404) return null; if (!response.ok) { @@ -261,6 +279,7 @@ export const defaultObjectStorage: CleanupObjectStorage = { accessKey, secretKey, ), + signal: requestSignal(), }); if (!response.ok) { throw new Error( @@ -287,6 +306,7 @@ export const defaultObjectStorage: CleanupObjectStorage = { accessKey, secretKey, ), + signal: requestSignal(), }); if (!response.ok && response.status !== 404) { throw new Error( diff --git a/packages/rooms/package.json b/packages/rooms/package.json index 98b5e6f..434ce90 100644 --- a/packages/rooms/package.json +++ b/packages/rooms/package.json @@ -18,6 +18,7 @@ }, "dependencies": { "@muster/authz": "workspace:*", + "@muster/config": "workspace:*", "@muster/contracts": "workspace:*", "@muster/database": "workspace:*", "drizzle-orm": "0.45.2", diff --git a/packages/rooms/src/agent-direct-message.ts b/packages/rooms/src/agent-direct-message.ts new file mode 100644 index 0000000..5d301ad --- /dev/null +++ b/packages/rooms/src/agent-direct-message.ts @@ -0,0 +1,333 @@ +import { createHash } from "node:crypto"; +import { requireCapability, type AuthorisationSubject } from "@muster/authz"; +import { redactObservationText } from "@muster/config"; +import { + appendAuditEvent, + database, + newId, + schema, + writeOutbox, +} from "@muster/database"; +import { and, eq, gt, isNull, or } from "drizzle-orm"; + +type Db = ReturnType; + +export type DirectMessageInvocation = + | { + handled: true; + queued: false; + duplicate: false; + reason: "agent_count" | "agent_unavailable"; + } + | { + handled: true; + queued: true; + duplicate: boolean; + agentId: string; + agentRunId: string; + status: string; + }; + +export class AgentDirectMessageDomainService { + constructor(private readonly db: Db = database()) {} + + async maybeQueue( + subject: AuthorisationSubject, + input: { messageId: string; roomId: string }, + traceId: string, + ): Promise { + const [source] = await this.db + .select({ + plainText: schema.messages.plainText, + relatedInvestigationId: schema.messages.relatedInvestigationId, + }) + .from(schema.messages) + .innerJoin( + schema.rooms, + and( + eq(schema.rooms.organisationId, subject.organisationId), + eq(schema.rooms.id, schema.messages.roomId), + eq(schema.rooms.roomType, "direct"), + isNull(schema.rooms.archivedAt), + ), + ) + .innerJoin( + schema.roomMemberships, + and( + eq(schema.roomMemberships.organisationId, subject.organisationId), + eq(schema.roomMemberships.roomId, schema.messages.roomId), + eq(schema.roomMemberships.actorId, subject.actorId), + or( + isNull(schema.roomMemberships.accessExpiresAt), + gt(schema.roomMemberships.accessExpiresAt, new Date()), + ), + ), + ) + .innerJoin( + schema.actors, + and( + eq(schema.actors.organisationId, subject.organisationId), + eq(schema.actors.id, schema.messages.authorActorId), + eq(schema.actors.id, subject.actorId), + eq(schema.actors.actorType, "human"), + eq(schema.actors.status, "active"), + ), + ) + .where( + and( + eq(schema.messages.organisationId, subject.organisationId), + eq(schema.messages.id, input.messageId), + eq(schema.messages.roomId, input.roomId), + eq(schema.messages.messageType, "text"), + isNull(schema.messages.deletedAt), + ), + ) + .limit(1); + if (!source) return null; + + const members = await this.db + .select({ + actorId: schema.actors.id, + actorStatus: schema.actors.status, + definitionId: schema.agentDefinitions.id, + definitionStatus: schema.agentDefinitions.status, + killSwitch: schema.agentDefinitions.killSwitch, + allowedRooms: schema.agentDefinitions.allowedRooms, + runtime: schema.agentDefinitions.runtime, + model: schema.agentDefinitions.model, + promptVersion: schema.agentDefinitions.systemPromptVersion, + maximumRuntimeSeconds: schema.agentDefinitions.maximumRuntimeSeconds, + maximumTokenBudget: schema.agentDefinitions.maximumTokenBudget, + maximumCostCents: schema.agentDefinitions.maximumCostCents, + }) + .from(schema.roomMemberships) + .innerJoin( + schema.actors, + and( + eq(schema.actors.organisationId, subject.organisationId), + eq(schema.actors.id, schema.roomMemberships.actorId), + eq(schema.actors.actorType, "agent"), + ), + ) + .leftJoin( + schema.agentDefinitions, + and( + eq(schema.agentDefinitions.organisationId, subject.organisationId), + eq(schema.agentDefinitions.id, schema.actors.id), + ), + ) + .where( + and( + eq(schema.roomMemberships.organisationId, subject.organisationId), + eq(schema.roomMemberships.roomId, input.roomId), + or( + isNull(schema.roomMemberships.accessExpiresAt), + gt(schema.roomMemberships.accessExpiresAt, new Date()), + ), + ), + ); + if (members.length === 0) return null; + + // Capability check before eligibility outcomes so unauthorised callers + // cannot fingerprint agent membership or readiness via reason codes. + requireCapability(subject, "agents.invoke"); + + const active = members.filter( + (member) => + member.actorStatus === "active" && + member.definitionId !== null && + member.definitionStatus === "active" && + member.killSwitch === false, + ); + if (active.length !== 1) { + return { + handled: true, + queued: false, + duplicate: false, + reason: "agent_count", + }; + } + const agent = active[0]!; + if ( + !Array.isArray(agent.allowedRooms) || + !agent.allowedRooms.includes(input.roomId) || + !agent.definitionId || + !agent.runtime || + !agent.model || + !agent.promptVersion || + agent.maximumRuntimeSeconds === null || + agent.maximumTokenBudget === null || + agent.maximumCostCents === null + ) { + return { + handled: true, + queued: false, + duplicate: false, + reason: "agent_unavailable", + }; + } + + const agentId = agent.definitionId; + const runtime = agent.runtime; + const model = agent.model; + const promptVersion = agent.promptVersion; + const maximumRuntimeSeconds = agent.maximumRuntimeSeconds; + const maximumTokenBudget = agent.maximumTokenBudget; + const maximumCostCents = agent.maximumCostCents; + const idempotencyKey = `agent-direct-message:${input.messageId}`; + const inputHash = createHash("sha256") + .update( + JSON.stringify({ + messageId: input.messageId, + roomId: input.roomId, + plainText: source.plainText, + }), + ) + .digest("hex"); + const deadlineAt = new Date(Date.now() + maximumRuntimeSeconds * 1_000); + + return this.db.transaction(async (tx) => { + const eligible = await tx + .select({ + id: schema.agentDefinitions.id, + allowedRooms: schema.agentDefinitions.allowedRooms, + }) + .from(schema.agentDefinitions) + .innerJoin( + schema.actors, + and( + eq(schema.actors.organisationId, subject.organisationId), + eq(schema.actors.id, schema.agentDefinitions.id), + eq(schema.actors.status, "active"), + ), + ) + .innerJoin( + schema.roomMemberships, + and( + eq(schema.roomMemberships.organisationId, subject.organisationId), + eq(schema.roomMemberships.roomId, input.roomId), + eq(schema.roomMemberships.actorId, schema.agentDefinitions.id), + or( + isNull(schema.roomMemberships.accessExpiresAt), + gt(schema.roomMemberships.accessExpiresAt, new Date()), + ), + ), + ) + .where( + and( + eq(schema.agentDefinitions.organisationId, subject.organisationId), + eq(schema.agentDefinitions.status, "active"), + eq(schema.agentDefinitions.killSwitch, false), + ), + ); + if ( + eligible.length !== 1 || + eligible[0]?.id !== agentId || + !Array.isArray(eligible[0].allowedRooms) || + !eligible[0].allowedRooms.includes(input.roomId) + ) { + throw new Error("Direct-message agent is unavailable"); + } + + const [inserted] = await tx + .insert(schema.agentRuns) + .values({ + id: newId(), + agentId, + organisationId: subject.organisationId, + roomId: input.roomId, + investigationId: source.relatedInvestigationId, + requestedByActorId: subject.actorId, + trigger: "direct_message", + status: "queued", + request: { + kind: "direct_message", + sourceMessageId: input.messageId, + humanRequest: source.plainText, + traceId, + }, + progress: { stage: "queued", percent: 0 }, + deadlineAt, + inputHash, + promptVersion, + runtime, + model, + maximumRuntimeSeconds, + maximumTokenBudget, + maximumCostCents, + idempotencyKey, + }) + .onConflictDoNothing() + .returning(); + const run = + inserted ?? + ( + await tx + .select() + .from(schema.agentRuns) + .where( + and( + eq(schema.agentRuns.organisationId, subject.organisationId), + eq(schema.agentRuns.idempotencyKey, idempotencyKey), + ), + ) + .limit(1) + )[0]; + if ( + !run || + run.agentId !== agentId || + run.roomId !== input.roomId || + run.requestedByActorId !== subject.actorId || + run.inputHash !== inputHash + ) { + throw new Error("Direct-message run idempotency conflict"); + } + if (inserted) { + await tx.insert(schema.agentRunEvents).values({ + id: newId(), + organisationId: subject.organisationId, + runId: run.id, + eventType: "queued", + message: "Direct-message agent run queued", + payload: { + trigger: "direct_message", + sourceMessageId: input.messageId, + roomId: input.roomId, + }, + }); + await appendAuditEvent(tx, { + organisationId: subject.organisationId, + actorId: subject.actorId, + actorType: "human", + action: "agent.run.queued", + targetType: "agent_run", + targetId: run.id, + metadata: { + trigger: "direct_message", + sourceMessageId: input.messageId, + roomId: input.roomId, + }, + traceId: redactObservationText(traceId), + }); + await writeOutbox(tx, { + organisationId: subject.organisationId, + eventType: "agent.run.queued", + aggregateType: "agent_run", + aggregateId: run.id, + queueName: "muster-agents", + payload: { runId: run.id }, + idempotencyKey: `agent.run.queued:${run.id}`, + traceId: redactObservationText(traceId), + }); + } + return { + handled: true, + queued: true, + duplicate: !inserted, + agentId: run.agentId, + agentRunId: run.id, + status: run.status, + }; + }); + } +} diff --git a/packages/rooms/src/index.ts b/packages/rooms/src/index.ts index 2891e60..51a2f5b 100644 --- a/packages/rooms/src/index.ts +++ b/packages/rooms/src/index.ts @@ -699,6 +699,7 @@ export class RoomService { membershipRole: schema.roomMemberships.membershipRole, policies: schema.rooms.policies, archivedAt: schema.rooms.archivedAt, + roomType: schema.rooms.roomType, }) .from(schema.roomMemberships) .innerJoin( @@ -929,6 +930,28 @@ export class RoomService { idempotencyKey: `${eventType}:${parsed.idempotencyKey}`, traceId, }); + // Durable direct-message agent evaluation in the same transaction as the + // message so a crash after commit still redrives invocation via outbox. + if ( + membership.roomType === "direct" && + !parsed.threadParentId && + parsed.messageType === "text" + ) { + await writeOutbox(tx, { + organisationId: subject.organisationId, + eventType: "agent.direct_message.evaluate", + aggregateType: "message", + aggregateId: id, + queueName: "muster-agents", + payload: { + messageId: id, + roomId: parsed.roomId, + actorId: subject.actorId, + }, + idempotencyKey: `agent.direct_message.evaluate:${id}`, + traceId, + }); + } await appendAuditEvent(tx, { organisationId: subject.organisationId, actorId: subject.actorId, @@ -1487,3 +1510,7 @@ export class RoomService { } export * from "./governance.ts"; +export { + AgentDirectMessageDomainService, + type DirectMessageInvocation, +} from "./agent-direct-message.ts"; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 383ea9a..81c90dd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -254,6 +254,9 @@ importers: '@muster/agents': specifier: workspace:* version: link:../../packages/agents + '@muster/authz': + specifier: workspace:* + version: link:../../packages/authz '@muster/config': specifier: workspace:* version: link:../../packages/config @@ -269,6 +272,9 @@ importers: '@muster/integrations': specifier: workspace:* version: link:../../packages/integrations + '@muster/rooms': + specifier: workspace:* + version: link:../../packages/rooms bullmq: specifier: 5.81.2 version: 5.81.2 @@ -656,6 +662,9 @@ importers: '@muster/authz': specifier: workspace:* version: link:../authz + '@muster/config': + specifier: workspace:* + version: link:../config '@muster/contracts': specifier: workspace:* version: link:../contracts