diff --git a/.env.example b/.env.example index d401e5b..acbe41f 100644 --- a/.env.example +++ b/.env.example @@ -30,3 +30,6 @@ SENTINEL_CLIENT_ID= SENTINEL_CLIENT_SECRET= SENTINEL_WORKSPACE_ID= MUSTER_MOCK_INTEGRATIONS=true +MUSTER_AGENT_GATEWAY_TOKEN=replace-with-at-least-32-random-bytes +# Optional comma-separated HTTPS origins for additional Alfie research feeds. +MUSTER_RESEARCH_ALLOWED_FEED_ORIGINS= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index be85dc5..d344f86 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -76,6 +76,8 @@ jobs: mc version enable ci/muster-evidence && mc anonymous set none ci/muster-evidence' - run: pnpm install --frozen-lockfile + - run: pnpm test:release-homelab + - run: pnpm test:release-image - name: Build database dependencies run: pnpm exec turbo build --filter=@muster/database - run: pnpm db:migrate @@ -103,8 +105,18 @@ jobs: if-no-files-found: ignore retention-days: 14 + release-security: + permissions: + contents: read + security-events: write + uses: ./.github/workflows/security.yml + container: + needs: quality runs-on: ubuntu-24.04 + outputs: + image_ref: ${{ steps.image.outputs.ref }} + image_digest: ${{ steps.build.outputs.digest }} permissions: contents: read packages: write @@ -112,10 +124,23 @@ jobs: id-token: write env: REGISTRY: ghcr.io - IMAGE_NAME: ${{ github.repository }} steps: - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3 + - name: Normalize image reference + id: image + shell: bash + run: | + image_ref="${REGISTRY}/${GITHUB_REPOSITORY,,}" + if [[ "$GITHUB_EVENT_NAME" == "push" ]]; then + build_ref="${image_ref}:staging-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" + else + build_ref="${image_ref}:verify-${GITHUB_SHA}" + fi + { + printf 'ref=%s\n' "$image_ref" + printf 'build_ref=%s\n' "$build_ref" + } >> "$GITHUB_OUTPUT" - name: Log in to GitHub Container Registry if: github.event_name == 'push' uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3 @@ -123,19 +148,6 @@ jobs: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - - name: Generate image tags and OCI labels - id: meta - uses: docker/metadata-action@c299e40c65443455700f0fdfc63efafe5b349051 # v5 - with: - images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} - tags: | - type=raw,value=latest,enable={{is_default_branch}} - type=ref,event=tag - type=sha - labels: | - org.opencontainers.image.title=Muster - org.opencontainers.image.description=Shared workspace for human and agent-driven security operations - org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }} - uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0 id: build with: @@ -143,37 +155,149 @@ jobs: push: ${{ github.event_name == 'push' }} load: ${{ github.event_name == 'pull_request' }} platforms: linux/amd64 - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - provenance: ${{ github.event_name == 'push' && 'mode=max' || 'false' }} - sbom: ${{ github.event_name == 'push' }} + tags: ${{ steps.image.outputs.build_ref }} + labels: | + org.opencontainers.image.title=Muster + org.opencontainers.image.description=Shared workspace for human and agent-driven security operations + org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }} + provenance: mode=max + sbom: true cache-from: type=gha cache-to: type=gha,mode=max - - name: Verify container starts + - name: Scan built image with Trivy + run: | + docker run --rm \ + --volume /var/run/docker.sock:/var/run/docker.sock \ + aquasec/trivy:0.67.2 image \ + --exit-code 1 --ignore-unfixed --severity HIGH,CRITICAL \ + "${{ steps.image.outputs.build_ref }}" + - name: Generate release SBOM + if: github.event_name == 'push' + uses: anchore/sbom-action@e22c389904149dbc22b58101806040fa8d37a610 # v0 + with: + image: ${{ steps.image.outputs.ref }}@${{ steps.build.outputs.digest }} + format: cyclonedx-json + output-file: muster-sbom.cdx.json + upload-artifact: false + - name: Generate pull request SBOM if: github.event_name == 'pull_request' - shell: bash + uses: anchore/sbom-action@e22c389904149dbc22b58101806040fa8d37a610 # v0 + with: + image: ${{ steps.image.outputs.build_ref }} + format: cyclonedx-json + output-file: muster-sbom.cdx.json + upload-artifact: false + - name: Verify staged OCI application platform + if: github.event_name == 'push' + run: | + docker buildx imagetools inspect \ + "${{ steps.image.outputs.ref }}@${{ steps.build.outputs.digest }}" --raw | + ./scripts/verify-image-platform.sh + - name: Record immutable image evidence and checksums + if: github.event_name == 'push' run: | - image_ref="${REGISTRY}/${IMAGE_NAME,,}:sha-${GITHUB_SHA::7}" - docker image inspect "$image_ref" - - name: Attest published image + printf '%s\n' \ + "${{ steps.image.outputs.ref }}@${{ steps.build.outputs.digest }}" > muster-image.txt + sha256sum muster-sbom.cdx.json muster-image.txt > SHA256SUMS + - name: Upload release evidence + if: github.event_name == 'push' + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: muster-release-evidence + path: | + muster-sbom.cdx.json + muster-image.txt + SHA256SUMS + - name: Verify container starts + if: github.event_name == 'pull_request' + run: docker image inspect "${{ steps.image.outputs.build_ref }}" + - name: Attest published image provenance + if: github.event_name == 'push' + uses: actions/attest@f7c74d28b9d84cb8768d0b8ca14a4bac6ef463e6 # v4 + with: + subject-name: ${{ steps.image.outputs.ref }} + subject-digest: ${{ steps.build.outputs.digest }} + push-to-registry: true + - name: Attest published image SBOM if: github.event_name == 'push' uses: actions/attest@f7c74d28b9d84cb8768d0b8ca14a4bac6ef463e6 # v4 with: - subject-name: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + subject-name: ${{ steps.image.outputs.ref }} subject-digest: ${{ steps.build.outputs.digest }} + sbom-path: muster-sbom.cdx.json push-to-registry: true - - name: Verify anonymous public pull + - name: Verify anonymous staging pull if: github.event_name == 'push' shell: bash run: | - public_image="${REGISTRY}/${IMAGE_NAME,,}:sha-${GITHUB_SHA::7}" - docker logout "$REGISTRY" + public_image="${{ steps.image.outputs.build_ref }}" + docker logout "$REGISTRY" || true docker pull "$public_image" + docker image inspect "$public_image" \ + --format '{{.Os}}/{{.Architecture}}' | grep -x 'linux/amd64' + docker buildx imagetools inspect "$public_image" --raw | + ./scripts/verify-image-platform.sh + + promote: + if: github.event_name == 'push' + needs: [container, release-security] + runs-on: ubuntu-24.04 + concurrency: + group: muster-release-tags-${{ github.repository }} + cancel-in-progress: false + permissions: + contents: read + packages: write + env: + REGISTRY: ghcr.io + IMAGE_REF: ${{ needs.container.outputs.image_ref }} + IMAGE_DIGEST: ${{ needs.container.outputs.image_digest }} + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + - uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3 + - name: Generate release tags + id: meta + uses: docker/metadata-action@c299e40c65443455700f0fdfc63efafe5b349051 # v5 + with: + images: ${{ env.IMAGE_REF }} + tags: | + type=ref,event=tag + type=sha,format=long + - name: Log in to GitHub Container Registry + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - name: Promote verified digest to release tags + env: + RELEASE_TAGS: ${{ steps.meta.outputs.tags }} + run: | + promotion_policy() { + case "${1##*:}" in + "sha-${GITHUB_SHA}" | v*) printf 'immutable\n' ;; + *) + printf 'Unexpected release tag: %s\n' "$1" >&2 + return 1 + ;; + esac + } + while IFS= read -r tag; do + [[ -n "$tag" ]] || continue + policy="$(promotion_policy "$tag")" + ./scripts/promote-image-tag.sh \ + "$tag" "${IMAGE_REF}@${IMAGE_DIGEST}" "$policy" check + done <<< "$RELEASE_TAGS" + while IFS= read -r tag; do + [[ -n "$tag" ]] || continue + policy="$(promotion_policy "$tag")" + ./scripts/promote-image-tag.sh \ + "$tag" "${IMAGE_REF}@${IMAGE_DIGEST}" "$policy" apply + done <<< "$RELEASE_TAGS" - name: Publication summary - if: github.event_name == 'push' run: | { echo "### Published container" - echo "\`${REGISTRY}/${IMAGE_NAME}\`" + echo "\`${IMAGE_REF}@${IMAGE_DIGEST}\`" echo "The workflow verified an anonymous pull after publication. GHCR visibility is configured once at the package level and retained by subsequent releases." } >> "$GITHUB_STEP_SUMMARY" diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index d4217f6..13febb7 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -1,9 +1,7 @@ name: Security and supply chain on: - push: - branches: [main] - pull_request: + workflow_call: schedule: - cron: "17 3 * * 1" @@ -66,6 +64,7 @@ jobs: output: trivy.sarif severity: HIGH,CRITICAL ignore-unfixed: true + scanners: vuln exit-code: "1" - uses: anchore/sbom-action@e22c389904149dbc22b58101806040fa8d37a610 # v0 with: diff --git a/README.md b/README.md index 227752e..7155219 100644 --- a/README.md +++ b/README.md @@ -177,8 +177,10 @@ that will use the service. When using HTTPS behind a reverse proxy, set `AUTH_SECURE_COOKIES=true`; do not add broad wildcard origins. The installer creates `.env.homelab` with mode `600`. Keep it out of source -control. The generated topology still uses synthetic connector endpoints until -you deliberately configure governed real connectors. +control. Its generated `MUSTER_AGENT_GATEWAY_TOKEN` authenticates internal +web/worker calls to the unexposed agent gateway; do not reuse or publish it. +The generated topology still uses synthetic connector endpoints until you +deliberately configure governed real connectors. ### Codex subscription authentication diff --git a/apps/agent-gateway/package.json b/apps/agent-gateway/package.json index 9d03506..b9eac69 100644 --- a/apps/agent-gateway/package.json +++ b/apps/agent-gateway/package.json @@ -16,6 +16,7 @@ "@muster/config": "workspace:*", "@muster/contracts": "workspace:*", "@muster/database": "workspace:*", + "@muster/integrations": "workspace:*", "@openai/codex": "0.145.0", "@openai/codex-sdk": "0.145.0", "drizzle-orm": "0.45.2", diff --git a/apps/agent-gateway/src/index.ts b/apps/agent-gateway/src/index.ts index 9762b29..3a5778b 100644 --- a/apps/agent-gateway/src/index.ts +++ b/apps/agent-gateway/src/index.ts @@ -15,6 +15,10 @@ import { import { and, eq } from "drizzle-orm"; import { z } from "zod"; import { DurableAgentRuntime } from "./runtime.ts"; +import { + isGatewayRequestAuthorised, + parseGatewayOrganisationId, +} from "./service-auth.ts"; const AgentRunRequestSchema = AgentInvestigationJobSchema.extend({ humanRequest: z.string().trim().min(1).max(4_000).optional(), @@ -24,6 +28,10 @@ const executionRuntime = process.env.MUSTER_AGENT_RUNTIME === "mock" ? "mock" : "codex"; const codexHome = process.env.CODEX_HOME ?? "/var/lib/muster/codex"; const globalKillSwitch = process.env.AGENT_KILL_SWITCH === "true"; +const gatewayToken = z + .string() + .min(32) + .parse(process.env.MUSTER_AGENT_GATEWAY_TOKEN); const runtime = new DurableAgentRuntime({ executionRuntime, codexHome, @@ -47,6 +55,12 @@ async function body(request: IncomingMessage): Promise { return JSON.parse(Buffer.concat(chunks).toString("utf8")); } +function requestOrganisationId(request: IncomingMessage) { + return parseGatewayOrganisationId( + request.headers["x-muster-organisation-id"], + ); +} + async function queueDirectRun( input: z.infer, idempotencyKey: string, @@ -170,12 +184,26 @@ const server = createServer(async (incoming, response) => { return; } + if ( + !isGatewayRequestAuthorised(incoming.headers.authorization, gatewayToken) + ) { + response.writeHead(401); + response.end(JSON.stringify({ error: "Unauthorised" })); + return; + } + const runMatch = incoming.method === "GET" ? url.pathname.match(/^\/v1\/runs\/([^/]+)$/) : null; if (runMatch?.[1]) { - const run = await runtime.read(runMatch[1]); + const organisationId = requestOrganisationId(incoming); + if (!organisationId) { + response.writeHead(400); + response.end(JSON.stringify({ error: "Organisation header required" })); + return; + } + const run = await runtime.read(runMatch[1], organisationId); response.writeHead(run ? 200 : 404); response.end(JSON.stringify(run ?? { error: "Run not found" })); return; @@ -234,6 +262,12 @@ const server = createServer(async (incoming, response) => { ); return; } + const organisationId = requestOrganisationId(incoming); + if (!organisationId || organisationId !== parsed.data.organisationId) { + response.writeHead(403); + response.end(JSON.stringify({ error: "Organisation mismatch" })); + return; + } const idempotencyKey = incoming.headers["idempotency-key"]?.toString().trim() || `api:${parsed.data.traceId}`; @@ -269,7 +303,13 @@ const server = createServer(async (incoming, response) => { ? url.pathname.match(/^\/v1\/runs\/([^/]+)\/cancel$/) : null; if (cancelMatch?.[1]) { - const cancelled = await runtime.cancel(cancelMatch[1]); + const organisationId = requestOrganisationId(incoming); + if (!organisationId) { + response.writeHead(400); + response.end(JSON.stringify({ error: "Organisation header required" })); + return; + } + const cancelled = await runtime.cancel(cancelMatch[1], organisationId); response.writeHead(cancelled ? 202 : 404); response.end( JSON.stringify({ diff --git a/apps/agent-gateway/src/runtime.integration.test.ts b/apps/agent-gateway/src/runtime.integration.test.ts index aeea457..8a4766b 100644 --- a/apps/agent-gateway/src/runtime.integration.test.ts +++ b/apps/agent-gateway/src/runtime.integration.test.ts @@ -18,13 +18,10 @@ const describeIntegration = integration ? describe.sequential : describe.skip; describe("Codex structured output schema", () => { it("removes unsupported URI formats while preserving authoritative validation", () => { - const generated = z.toJSONSchema( - AgentStructuredOutputSchemas.HuntResult, - { - target: "draft-2020-12", - io: "output", - }, - ); + const generated = z.toJSONSchema(AgentStructuredOutputSchemas.HuntResult, { + target: "draft-2020-12", + io: "output", + }); expect(JSON.stringify(generated)).toContain('"format":"uri"'); expect(JSON.stringify(codexOutputSchemaFor("HuntResult"))).not.toContain( '"format":"uri"', @@ -172,6 +169,42 @@ describeIntegration("durable agent runtime", () => { throw new Error(`Run ${runId} did not reach ${status}`); } + async function directMessageSource(suffix: string) { + const [room] = await database() + .select({ id: schema.rooms.id }) + .from(schema.rooms) + .innerJoin( + schema.roomMemberships, + and( + eq(schema.roomMemberships.organisationId, organisationId), + eq(schema.roomMemberships.roomId, schema.rooms.id), + eq(schema.roomMemberships.actorId, agentId), + ), + ) + .where( + and( + eq(schema.rooms.organisationId, organisationId), + eq(schema.rooms.roomType, "direct"), + ), + ) + .limit(1); + if (!room) throw new Error("Seeded agent direct room required"); + const messageId = newId(); + await database() + .insert(schema.messages) + .values({ + id: messageId, + organisationId, + roomId: room.id, + authorActorId: requestedByActorId, + messageType: "text", + document: { type: "doc", content: [] }, + plainText: `Synthetic direct request ${suffix}`, + idempotencyKey: `runtime-direct-source:${messageId}`, + }); + return { messageId, roomId: room.id }; + } + it("recovers an expired lease without duplicating the run", async () => { const run = await insertRun("restart"); const duplicateId = newId(); @@ -229,7 +262,11 @@ describeIntegration("durable agent runtime", () => { codexHome: "/tmp/muster-runtime-integration", }); expect( - await runtime.cancel(run.id, "Synthetic operator cancellation"), + await runtime.cancel( + run.id, + organisationId, + "Synthetic operator cancellation", + ), ).toBe(true); const cancelled = await waitFor(run.id, "cancelled"); expect(cancelled.cancellationRequestedAt).not.toBeNull(); @@ -238,6 +275,22 @@ describeIntegration("durable agent runtime", () => { ); }); + it("scopes run reads and cancellations by organisation", async () => { + const run = await insertRun("cross-organisation-guard"); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + }); + const otherOrganisationId = newId(); + + await expect(runtime.read(run.id, otherOrganisationId)).resolves.toBeNull(); + await expect(runtime.cancel(run.id, otherOrganisationId)).resolves.toBe( + false, + ); + const persisted = await waitFor(run.id, "queued"); + expect(persisted.organisationId).toBe(organisationId); + }); + it("returns a redacted observer projection without changing the execution record", async () => { const canary = `synthetic-api-secret-${newId()}`; const run = await insertRun("redacted-observer", { @@ -269,7 +322,7 @@ describeIntegration("durable agent runtime", () => { codexHome: "/tmp/muster-runtime-integration", }); - const projection = await runtime.read(run.id); + const projection = await runtime.read(run.id, organisationId); const serialisedProjection = JSON.stringify(projection); expect(serialisedProjection).not.toContain(canary); expect(serialisedProjection).toContain("[REDACTED]"); @@ -577,4 +630,310 @@ describeIntegration("durable agent runtime", () => { }); expect(message?.relatedAgentRunId).toBe(run.id); }); + + it("projects a completed direct-message run as one linked room reply", async () => { + const source = await directMessageSource("completed"); + const run = await insertRun("direct-completed", { + roomId: source.roomId, + request: { + kind: "direct_message", + sourceMessageId: source.messageId, + humanRequest: "Review the synthetic direct request", + traceId: `integration-direct-${source.messageId}`, + }, + }); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + mockDelayMs: 10, + }); + await runtime.dispatch(); + await waitFor(run.id, "completed"); + runtime.stop(); + + const [replies, outbox] = await Promise.all([ + database() + .select() + .from(schema.messages) + .where( + eq( + schema.messages.idempotencyKey, + `agent-direct-message-reply:${run.id}`, + ), + ), + database() + .select() + .from(schema.outboxEvents) + .where( + eq( + schema.outboxEvents.idempotencyKey, + `room.message.created:agent-direct-message:${run.id}`, + ), + ), + ]); + expect(replies).toHaveLength(1); + expect(outbox).toHaveLength(1); + const reply = replies[0]; + expect(reply).toMatchObject({ + roomId: source.roomId, + threadParentId: source.messageId, + authorActorId: agentId, + messageType: "agent-status", + relatedAgentRunId: run.id, + }); + expect(reply?.document).toMatchObject({ + type: "agent-direct-message-reply", + status: "completed", + sourceMessageId: source.messageId, + agentRunId: run.id, + trust: "agent-analysis", + }); + expect(outbox[0]?.aggregateId).toBe(reply?.id); + }); + + it("projects a failed direct-message run as one linked room reply", async () => { + const source = await directMessageSource("failed"); + const run = await insertRun("direct-failed", { + roomId: source.roomId, + maximumTokenBudget: 1, + request: { + kind: "direct_message", + sourceMessageId: source.messageId, + humanRequest: "Review the synthetic direct request", + traceId: `integration-direct-${source.messageId}`, + }, + }); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + mockDelayMs: 10, + }); + await runtime.dispatch(); + await waitFor(run.id, "failed"); + runtime.stop(); + + const replies = await database() + .select() + .from(schema.messages) + .where( + eq( + schema.messages.idempotencyKey, + `agent-direct-message-reply:${run.id}`, + ), + ); + expect(replies).toHaveLength(1); + expect(replies[0]).toMatchObject({ + roomId: source.roomId, + threadParentId: source.messageId, + authorActorId: agentId, + messageType: "agent-status", + relatedAgentRunId: run.id, + }); + expect(replies[0]?.document).toMatchObject({ + type: "agent-direct-message-reply", + status: "failed", + sourceMessageId: source.messageId, + agentRunId: run.id, + failureCode: "token_ceiling", + }); + const outbox = await database() + .select() + .from(schema.outboxEvents) + .where( + eq( + schema.outboxEvents.idempotencyKey, + `room.message.created:agent-direct-message:${run.id}`, + ), + ); + expect(outbox).toHaveLength(1); + }); + + it("projects a direct-message kill-switch failure before execution", async () => { + const source = await directMessageSource("kill-switch"); + const run = await insertRun("direct-kill-switch", { + roomId: source.roomId, + request: { + kind: "direct_message", + sourceMessageId: source.messageId, + humanRequest: "Review the synthetic direct request", + traceId: `integration-direct-${source.messageId}`, + }, + }); + await database() + .update(schema.agentDefinitions) + .set({ killSwitch: true }) + .where(eq(schema.agentDefinitions.id, agentId)); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + mockDelayMs: 10, + }); + try { + await runtime.dispatch(); + await waitFor(run.id, "failed"); + } finally { + runtime.stop(); + await database() + .update(schema.agentDefinitions) + .set({ killSwitch: false }) + .where(eq(schema.agentDefinitions.id, agentId)); + } + + const [reply] = await database() + .select() + .from(schema.messages) + .where( + eq( + schema.messages.idempotencyKey, + `agent-direct-message-reply:${run.id}`, + ), + ); + expect(reply).toMatchObject({ + threadParentId: source.messageId, + messageType: "agent-status", + relatedAgentRunId: run.id, + }); + expect(reply?.document).toMatchObject({ + status: "failed", + failureCode: "agent_kill_switch", + sourceMessageId: source.messageId, + agentRunId: run.id, + }); + }); + + it("projects a cancelled direct-message run exactly once", async () => { + const source = await directMessageSource("cancelled"); + const run = await insertRun("direct-cancelled", { + roomId: source.roomId, + request: { + kind: "direct_message", + sourceMessageId: source.messageId, + humanRequest: "Review the synthetic direct request", + traceId: `integration-direct-${source.messageId}`, + }, + }); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + mockDelayMs: 10, + }); + expect( + await runtime.cancel( + run.id, + organisationId, + "Synthetic operator cancellation", + ), + ).toBe(true); + await waitFor(run.id, "cancelled"); + runtime.stop(); + + const [replies, outbox] = await Promise.all([ + database() + .select() + .from(schema.messages) + .where( + eq( + schema.messages.idempotencyKey, + `agent-direct-message-reply:${run.id}`, + ), + ), + database() + .select() + .from(schema.outboxEvents) + .where( + eq( + schema.outboxEvents.idempotencyKey, + `room.message.created:agent-direct-message:${run.id}`, + ), + ), + ]); + expect(replies).toHaveLength(1); + expect(replies[0]).toMatchObject({ + roomId: source.roomId, + threadParentId: source.messageId, + authorActorId: agentId, + messageType: "agent-status", + relatedAgentRunId: run.id, + }); + expect(replies[0]?.document).toMatchObject({ + status: "cancelled", + failureCode: "operator_cancelled", + sourceMessageId: source.messageId, + agentRunId: run.id, + }); + expect(outbox).toHaveLength(1); + }); + + it("fails queued direct messages when room authorisation is revoked", async () => { + const source = await directMessageSource("revoked-room"); + const run = await insertRun("direct-revoked-room", { + roomId: source.roomId, + request: { + kind: "direct_message", + sourceMessageId: source.messageId, + humanRequest: "Review the synthetic direct request", + traceId: `integration-direct-${source.messageId}`, + }, + }); + const [definition] = await database() + .select({ allowedRooms: schema.agentDefinitions.allowedRooms }) + .from(schema.agentDefinitions) + .where(eq(schema.agentDefinitions.id, agentId)) + .limit(1); + await database() + .update(schema.agentDefinitions) + .set({ allowedRooms: [] }) + .where(eq(schema.agentDefinitions.id, agentId)); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + mockDelayMs: 10, + }); + try { + await runtime.dispatch(); + const failed = await waitFor(run.id, "failed"); + expect(failed.failureCode).toBe("direct_message_not_authorised"); + expect(failed.startedAt).toBeNull(); + } finally { + runtime.stop(); + await database() + .update(schema.agentDefinitions) + .set({ allowedRooms: definition?.allowedRooms ?? [] }) + .where(eq(schema.agentDefinitions.id, agentId)); + } + }); + + it("reports inactive agents truthfully before execution", async () => { + const source = await directMessageSource("inactive"); + const run = await insertRun("direct-inactive", { + roomId: source.roomId, + request: { + kind: "direct_message", + sourceMessageId: source.messageId, + humanRequest: "Review the synthetic direct request", + traceId: `integration-direct-${source.messageId}`, + }, + }); + await database() + .update(schema.agentDefinitions) + .set({ status: "inactive" }) + .where(eq(schema.agentDefinitions.id, agentId)); + const runtime = new DurableAgentRuntime({ + executionRuntime: "mock", + codexHome: "/tmp/muster-runtime-integration", + mockDelayMs: 10, + }); + try { + await runtime.dispatch(); + const failed = await waitFor(run.id, "failed"); + expect(failed.failureCode).toBe("agent_inactive"); + expect(failed.startedAt).toBeNull(); + } finally { + runtime.stop(); + await database() + .update(schema.agentDefinitions) + .set({ status: "active" }) + .where(eq(schema.agentDefinitions.id, agentId)); + } + }); }); diff --git a/apps/agent-gateway/src/runtime.ts b/apps/agent-gateway/src/runtime.ts index 63c48b7..cf635e8 100644 --- a/apps/agent-gateway/src/runtime.ts +++ b/apps/agent-gateway/src/runtime.ts @@ -27,20 +27,185 @@ import { TenantRepository, writeOutbox, } from "@muster/database"; -import { and, asc, eq, inArray, isNull, lt, or, sql } from "drizzle-orm"; +import { + ConnectorConfigurationSchema, + GovernedConnectorError, + QueryTemplateSchema, + decryptConnectorAuth, + encryptConnectorPayload, + executeGovernedQuery, + redactUntrusted, + type ConnectorAuth, + type ConnectorConfiguration, + type QueryTemplate, +} from "@muster/integrations"; +import { and, asc, eq, gt, inArray, isNull, lt, or, sql } from "drizzle-orm"; import { z } from "zod"; type AgentRunRow = typeof schema.agentRuns.$inferSelect; type Context = Awaited>; +type Db = ReturnType; +type Tx = Parameters[0]>[0]; type PersistedRequest = { - kind?: "jessie_hunt" | undefined; + kind?: "jessie_hunt" | "direct_message" | undefined; huntId?: string | undefined; huntPlan?: unknown; humanRequest?: string | undefined; + sourceMessageId?: string | undefined; traceId?: string | undefined; + harness?: { + mode?: "slack" | "hermes" | "mcp" | "cli" | "http" | undefined; + }; +}; + +type LiveConnectorEvidence = { + queryRunId?: string; + source: string; + product: string; + templateKey: string; + status: "succeeded" | "failed" | "unavailable"; + result?: unknown; + responseMetadata?: unknown; + errorCode?: string; + errorMessage?: string; }; +function terminalSummary(output: unknown) { + if (!output || typeof output !== "object" || Array.isArray(output)) + return "The agent completed the request with schema-valid output."; + const record = output as Record; + for (const key of ["summary", "narrative", "headline", "title"]) { + const value = record[key]; + if (typeof value === "string" && value.trim()) + return value.trim().slice(0, 10_000); + } + return "The agent completed the request with schema-valid output."; +} + +async function projectDirectMessageTerminalReply( + tx: Tx, + run: AgentRunRow, + request: PersistedRequest, + terminal: + | { + status: "completed"; + output: unknown; + outputHash: string; + outputSchema: AgentStructuredOutputName; + } + | { + status: "failed"; + failureCode: string; + error: string; + } + | { + status: "cancelled"; + failureCode: string; + error: string; + }, +) { + if ( + request.kind !== "direct_message" || + !request.sourceMessageId || + !run.roomId + ) + return; + const [source] = await tx + .select({ id: schema.messages.id }) + .from(schema.messages) + .innerJoin( + schema.rooms, + and( + eq(schema.rooms.organisationId, run.organisationId), + eq(schema.rooms.id, run.roomId), + eq(schema.rooms.roomType, "direct"), + isNull(schema.rooms.archivedAt), + ), + ) + .innerJoin( + schema.roomMemberships, + and( + eq(schema.roomMemberships.organisationId, run.organisationId), + eq(schema.roomMemberships.roomId, run.roomId), + eq(schema.roomMemberships.actorId, run.agentId), + or( + isNull(schema.roomMemberships.accessExpiresAt), + gt(schema.roomMemberships.accessExpiresAt, new Date()), + ), + ), + ) + .where( + and( + eq(schema.messages.organisationId, run.organisationId), + eq(schema.messages.id, request.sourceMessageId), + eq(schema.messages.roomId, run.roomId), + isNull(schema.messages.deletedAt), + ), + ) + .limit(1); + if (!source) return; + + const completed = terminal.status === "completed"; + const plainText = completed + ? terminalSummary(terminal.output) + : terminal.status === "cancelled" + ? `The agent request was cancelled (${terminal.failureCode}).` + : `The agent could not complete this request (${terminal.failureCode}). Retry the request or contact an operator if the problem continues.`; + const messageId = newId(); + const [message] = await tx + .insert(schema.messages) + .values({ + id: messageId, + organisationId: run.organisationId, + roomId: run.roomId, + threadParentId: request.sourceMessageId, + authorActorId: run.agentId, + messageType: "agent-status", + document: completed + ? { + type: "agent-direct-message-reply", + status: terminal.status, + sourceMessageId: request.sourceMessageId, + agentRunId: run.id, + outputSchema: terminal.outputSchema, + outputHash: terminal.outputHash, + summary: plainText, + trust: "agent-analysis", + } + : { + type: "agent-direct-message-reply", + status: terminal.status, + sourceMessageId: request.sourceMessageId, + agentRunId: run.id, + failureCode: terminal.failureCode, + }, + plainText, + dataClassification: "internal", + relatedInvestigationId: run.investigationId, + relatedAgentRunId: run.id, + idempotencyKey: `agent-direct-message-reply:${run.id}`, + }) + .onConflictDoNothing() + .returning({ id: schema.messages.id }); + if (!message) return; + await writeOutbox(tx, { + organisationId: run.organisationId, + eventType: "room.message.created", + aggregateType: "message", + aggregateId: message.id, + queueName: "muster-outbox", + payload: { + messageId: message.id, + roomId: run.roomId, + threadParentId: request.sourceMessageId, + agentRunId: run.id, + }, + idempotencyKey: `room.message.created:agent-direct-message:${run.id}`, + traceId: redactObservationText(request.traceId ?? `agent-run-${run.id}`), + }); +} + function removeUnsupportedCodexSchemaFormats(value: unknown): unknown { if (Array.isArray(value)) return value.map(removeUnsupportedCodexSchemaFormats); @@ -319,12 +484,17 @@ export class DurableAgentRuntime { this.lastReadinessSnapshotAt = now.getTime(); } - async read(runId: string) { + async read(runId: string, organisationId: string) { const db = database(); const [run] = await db .select() .from(schema.agentRuns) - .where(eq(schema.agentRuns.id, runId)) + .where( + and( + eq(schema.agentRuns.organisationId, organisationId), + eq(schema.agentRuns.id, runId), + ), + ) .limit(1); if (!run) return null; const events = await db @@ -360,7 +530,11 @@ export class DurableAgentRuntime { return redactForObservation(projection) as typeof projection; } - async cancel(runId: string, reason = "Cancelled by operator") { + async cancel( + runId: string, + organisationId: string, + reason = "Cancelled by operator", + ) { const now = new Date(); const [run] = await database().transaction(async (tx) => { const [updated] = await tx @@ -376,6 +550,7 @@ export class DurableAgentRuntime { }) .where( and( + eq(schema.agentRuns.organisationId, organisationId), eq(schema.agentRuns.id, runId), or( eq(schema.agentRuns.status, "awaiting_approval"), @@ -407,6 +582,28 @@ export class DurableAgentRuntime { this.request(updated).traceId ?? `agent-run-${updated.id}`, ), }); + await projectDirectMessageTerminalReply( + tx, + updated, + this.request(updated), + { + status: "cancelled", + failureCode: "operator_cancelled", + error: reason, + }, + ); + await writeOutbox(tx, { + organisationId: updated.organisationId, + eventType: "agent.run.settled", + aggregateType: "agent_run", + aggregateId: updated.id, + queueName: "muster-notifications", + payload: { runId: updated.id, status: "cancelled" }, + idempotencyKey: `agent.run.settled:${updated.id}`, + traceId: redactObservationText( + this.request(updated).traceId ?? `agent-run-${updated.id}`, + ), + }); const [hunt] = await tx .update(schema.huntRuns) .set({ @@ -505,8 +702,20 @@ export class DurableAgentRuntime { const recovered = candidate.status === "running"; const [run] = await database().transaction(async (tx) => { const [definition] = await tx - .select({ killSwitch: schema.agentDefinitions.killSwitch }) + .select({ + status: schema.agentDefinitions.status, + killSwitch: schema.agentDefinitions.killSwitch, + allowedRooms: schema.agentDefinitions.allowedRooms, + actorStatus: schema.actors.status, + }) .from(schema.agentDefinitions) + .leftJoin( + schema.actors, + and( + eq(schema.actors.organisationId, candidate.organisationId), + eq(schema.actors.id, schema.agentDefinitions.id), + ), + ) .where( and( eq(schema.agentDefinitions.id, candidate.agentId), @@ -517,14 +726,92 @@ export class DurableAgentRuntime { ), ) .limit(1); - if (!definition || definition.killSwitch) { + const request = this.request(candidate); + let eligibilityFailure: { code: string; message: string } | undefined; + if (!definition) { + eligibilityFailure = { + code: "agent_unavailable", + message: "Agent definition is unavailable", + }; + } else if (definition.killSwitch) { + eligibilityFailure = { + code: "agent_kill_switch", + message: "Agent is disabled by its kill switch", + }; + } else if ( + definition.status !== "active" || + definition.actorStatus !== "active" + ) { + eligibilityFailure = { + code: "agent_inactive", + message: "Agent is inactive", + }; + } else if (request.kind === "direct_message") { + const sourceMessageId = request.sourceMessageId; + const roomId = candidate.roomId; + if ( + !sourceMessageId || + !roomId || + !Array.isArray(definition.allowedRooms) || + !definition.allowedRooms.includes(roomId) + ) { + eligibilityFailure = { + code: "direct_message_not_authorised", + message: "Direct-message room is no longer authorised", + }; + } else { + const [authorisedRoom] = await tx + .select({ id: schema.messages.id }) + .from(schema.messages) + .innerJoin( + schema.rooms, + and( + eq(schema.rooms.organisationId, candidate.organisationId), + eq(schema.rooms.id, roomId), + eq(schema.rooms.roomType, "direct"), + isNull(schema.rooms.archivedAt), + ), + ) + .innerJoin( + schema.roomMemberships, + and( + eq( + schema.roomMemberships.organisationId, + candidate.organisationId, + ), + eq(schema.roomMemberships.roomId, roomId), + eq(schema.roomMemberships.actorId, candidate.agentId), + or( + isNull(schema.roomMemberships.accessExpiresAt), + gt(schema.roomMemberships.accessExpiresAt, now), + ), + ), + ) + .where( + and( + eq(schema.messages.organisationId, candidate.organisationId), + eq(schema.messages.id, sourceMessageId), + eq(schema.messages.roomId, roomId), + isNull(schema.messages.deletedAt), + ), + ) + .limit(1); + if (!authorisedRoom) { + eligibilityFailure = { + code: "direct_message_not_authorised", + message: "Direct-message room is no longer authorised", + }; + } + } + } + if (eligibilityFailure) { const [disabled] = await tx .update(schema.agentRuns) .set({ status: "failed", completedAt: now, - failureCode: "agent_kill_switch", - error: "Agent is disabled by its kill switch", + failureCode: eligibilityFailure.code, + error: eligibilityFailure.message, leaseExpiresAt: null, }) .where( @@ -540,9 +827,31 @@ export class DurableAgentRuntime { organisationId: disabled.organisationId, runId: disabled.id, eventType: "failed", - message: "Agent kill switch blocked execution", - payload: { failureCode: "agent_kill_switch" }, + message: eligibilityFailure.message, + payload: { failureCode: eligibilityFailure.code }, + }); + await appendAuditEvent(tx, { + organisationId: disabled.organisationId, + actorId: disabled.agentId, + actorType: "agent", + action: "agent.run.failed", + targetType: "agent_run", + targetId: disabled.id, + metadata: { failureCode: eligibilityFailure.code }, + traceId: redactObservationText( + this.request(disabled).traceId ?? `agent-run-${disabled.id}`, + ), }); + await projectDirectMessageTerminalReply( + tx, + disabled, + this.request(disabled), + { + status: "failed", + failureCode: eligibilityFailure.code, + error: eligibilityFailure.message, + }, + ); } return []; } @@ -681,7 +990,7 @@ export class DurableAgentRuntime { ), ); } else if (controller.signal.aborted) { - if (!this.stopping) await this.cancel(run.id); + if (!this.stopping) await this.cancel(run.id, run.organisationId); } else { await this.fail( run, @@ -802,8 +1111,10 @@ export class DurableAgentRuntime { evidence: [ { type: "fixture", - reference: "https://www.cisa.gov/known-exploited-vulnerabilities-catalog", - sha256: "0000000000000000000000000000000000000000000000000000000000000000", + reference: + "https://www.cisa.gov/known-exploited-vulnerabilities-catalog", + sha256: + "0000000000000000000000000000000000000000000000000000000000000000", }, ], }, @@ -866,8 +1177,23 @@ export class DurableAgentRuntime { comparisonPeriod: null, }, filters: { organisationScoped: true }, - metricDefinitions: [{ key: "mtta", definition: "Synthetic", population: "Synthetic", exclusions: "Synthetic" }], - values: [{ key: "mtta", value: null, unit: "minutes", state: "unavailable", sampleSize: 0 }], + metricDefinitions: [ + { + key: "mtta", + definition: "Synthetic", + population: "Synthetic", + exclusions: "Synthetic", + }, + ], + values: [ + { + key: "mtta", + value: null, + unit: "minutes", + state: "unavailable", + sampleSize: 0, + }, + ], sourceReferences: [{ source: "synthetic", query: {} }], narrative: base.summary, caveats: ["Synthetic runtime output"], @@ -889,21 +1215,39 @@ export class DurableAgentRuntime { private async heartbeat(runId: string) { const now = new Date(); - const updated = await database() - .update(schema.agentRuns) - .set({ - heartbeatAt: now, - leaseExpiresAt: new Date(now.getTime() + this.leaseMs), - progress: { stage: "executing", percent: 50 }, - }) - .where( - and( - eq(schema.agentRuns.id, runId), - eq(schema.agentRuns.status, "running"), - eq(schema.agentRuns.workerId, this.workerId), - ), - ) - .returning({ id: schema.agentRuns.id }); + const updated = await database().transaction(async (tx) => { + const rows = await tx + .update(schema.agentRuns) + .set({ + heartbeatAt: now, + leaseExpiresAt: new Date(now.getTime() + this.leaseMs), + progress: { stage: "executing", percent: 50 }, + }) + .where( + and( + eq(schema.agentRuns.id, runId), + eq(schema.agentRuns.status, "running"), + eq(schema.agentRuns.workerId, this.workerId), + ), + ) + .returning({ + id: schema.agentRuns.id, + organisationId: schema.agentRuns.organisationId, + }); + const run = rows[0]; + if (run) + await writeOutbox(tx, { + organisationId: run.organisationId, + eventType: "agent.run.progress", + aggregateType: "agent_run", + aggregateId: run.id, + queueName: "muster-notifications", + payload: { runId: run.id, stage: "executing", percent: 50 }, + idempotencyKey: `agent.run.progress:${run.id}:executing`, + traceId: `agent-run-${run.id}`, + }); + return rows; + }); if (updated.length === 0) this.activeRuns.get(runId)?.abort(); } @@ -1092,6 +1436,24 @@ export class DurableAgentRuntime { this.request(run).traceId ?? `agent-run-${run.id}`, ), }); + await projectDirectMessageTerminalReply(tx, run, this.request(run), { + status: "completed", + output: result.output, + outputHash: result.outputHash, + outputSchema: result.schemaName, + }); + await writeOutbox(tx, { + organisationId: updated.organisationId, + eventType: "agent.run.settled", + aggregateType: "agent_run", + aggregateId: updated.id, + queueName: "muster-notifications", + payload: { runId: updated.id, status: "completed" }, + idempotencyKey: `agent.run.settled:${updated.id}`, + traceId: redactObservationText( + this.request(updated).traceId ?? `agent-run-${updated.id}`, + ), + }); const [hunt] = await tx .update(schema.huntRuns) .set({ @@ -1265,6 +1627,23 @@ export class DurableAgentRuntime { this.request(run).traceId ?? `agent-run-${run.id}`, ), }); + await projectDirectMessageTerminalReply(tx, run, this.request(run), { + status: "failed", + failureCode: failure.code, + error: failure.message, + }); + await writeOutbox(tx, { + organisationId: updated.organisationId, + eventType: "agent.run.settled", + aggregateType: "agent_run", + aggregateId: updated.id, + queueName: "muster-notifications", + payload: { runId: updated.id, status: "failed" }, + idempotencyKey: `agent.run.settled:${updated.id}`, + traceId: redactObservationText( + this.request(updated).traceId ?? `agent-run-${updated.id}`, + ), + }); const [hunt] = await tx .update(schema.huntRuns) .set({ @@ -1327,10 +1706,11 @@ export class DurableAgentRuntime { private request(run: AgentRunRow): PersistedRequest { const parsed = z .object({ - kind: z.literal("jessie_hunt").optional(), + kind: z.enum(["jessie_hunt", "direct_message"]).optional(), huntId: z.uuid().optional(), huntPlan: z.unknown().optional(), humanRequest: z.string().optional(), + sourceMessageId: z.uuid().optional(), traceId: z.string().optional(), }) .safeParse(run.request); @@ -1478,7 +1858,21 @@ async function loadAuthoritativeContext( ); if (request.kind === "jessie_hunt" && !hunt) throw new RunFailure("Hunt not found in organisation", "hunt_not_found"); - return { investigation, actor, alerts, findings, hunt, huntQueries }; + const liveConnectorEvidence = await loadLiveConnectorEvidence({ + db, + actor, + runId, + request, + }); + return { + investigation, + actor, + alerts, + findings, + hunt, + huntQueries, + liveConnectorEvidence, + }; } function outputSchemaFor( @@ -1500,6 +1894,483 @@ function outputSchemaFor( return "TriageRecommendation"; } +type LiveTemplateRow = { + integration: typeof schema.integrationRecords.$inferSelect; + template: typeof schema.integrationQueryTemplates.$inferSelect; + credential: typeof schema.integrationConnectorCredentials.$inferSelect; +}; + +function connectorFailure(error: unknown) { + if (error instanceof GovernedConnectorError) + return { + code: error.code, + message: redactObservationText(error.message), + }; + return { + code: "source_unavailable", + message: redactObservationText( + error instanceof Error ? error.message : "Connector query failed", + ), + }; +} + +async function executeLiveContextQuery(input: { + db: Db; + row: LiveTemplateRow; + actor: typeof schema.actors.$inferSelect; + runId: string; + traceId: string; + values: Record; + suffix?: string; +}): Promise { + const key = process.env.CONNECTOR_ENCRYPTION_KEY; + if (!key) + return { + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: input.row.template.templateKey, + status: "unavailable", + errorCode: "connector_encryption_unavailable", + errorMessage: "Connector encryption is not configured.", + }; + const definition = QueryTemplateSchema.parse(input.row.template.definition); + const capabilities = Array.isArray(input.actor.capabilityAssignments) + ? input.actor.capabilityAssignments + : []; + if (!capabilities.includes(definition.requiredCapability)) + return { + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: definition.key, + status: "unavailable", + errorCode: "capability_revoked", + errorMessage: "The agent lacks the connector read capability.", + }; + const auth: ConnectorAuth = decryptConnectorAuth( + input.row.credential.encryptedCredential, + key, + ); + const { authType: _storedAuthType, ...storedConfiguration } = input.row + .integration.configuration as Record; + const configuration: ConnectorConfiguration = + ConnectorConfigurationSchema.parse({ + ...storedConfiguration, + auth, + }); + const suffix = input.suffix ? `:${input.suffix}` : ""; + const idempotencyKey = + `agent-context:${input.runId}:${definition.key}${suffix}`.slice(0, 200); + const queryRunId = newId(); + const [inserted] = await input.db.transaction(async (tx) => { + const [created] = await tx + .insert(schema.integrationQueryRuns) + .values({ + id: queryRunId, + organisationId: input.actor.organisationId, + integrationId: input.row.integration.id, + templateId: input.row.template.id, + requestedByActorId: input.actor.id, + idempotencyKey, + traceId: input.traceId, + status: "running", + input: { envelope: encryptConnectorPayload(input.values, key) }, + requestMetadata: { + source: "agent-live-context", + agentRunId: input.runId, + templateKey: definition.key, + }, + startedAt: new Date(), + }) + .onConflictDoNothing() + .returning(); + if (created) { + await appendAuditEvent(tx, { + organisationId: input.actor.organisationId, + actorId: input.actor.id, + actorType: input.actor.actorType, + action: "connector.query.started", + targetType: "integration_query", + targetId: created.id, + metadata: { + source: "agent-live-context", + agentRunId: input.runId, + integrationId: input.row.integration.id, + templateKey: definition.key, + templateVersion: definition.version, + }, + traceId: input.traceId, + }); + await writeOutbox(tx, { + organisationId: input.actor.organisationId, + eventType: "connector.query.started", + aggregateType: "integration_query", + aggregateId: created.id, + queueName: "muster-outbox", + payload: { + queryRunId: created.id, + agentRunId: input.runId, + templateKey: definition.key, + }, + idempotencyKey: `connector.query.started:${created.id}`, + traceId: input.traceId, + }); + } + return [created] as const; + }); + const run = + inserted ?? + ( + await input.db + .select() + .from(schema.integrationQueryRuns) + .where( + and( + eq( + schema.integrationQueryRuns.organisationId, + input.actor.organisationId, + ), + eq(schema.integrationQueryRuns.idempotencyKey, idempotencyKey), + ), + ) + .limit(1) + )[0]; + if (!run) + return { + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: definition.key, + status: "unavailable", + errorCode: "query_state_unavailable", + errorMessage: "Connector query state could not be created.", + }; + if (run.status === "succeeded") + return { + queryRunId: run.id, + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: definition.key, + status: "succeeded", + result: run.result, + responseMetadata: run.responseMetadata, + }; + if (run.status === "failed") + return { + queryRunId: run.id, + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: definition.key, + status: "failed", + errorCode: run.errorCode ?? "source_unavailable", + errorMessage: run.errorMessage ?? "Connector query failed.", + }; + try { + const result = await executeGovernedQuery({ + configuration, + auth, + template: definition, + values: input.values, + }); + const safeResult = redactUntrusted(result.data); + await input.db.transaction(async (tx) => { + await tx + .update(schema.integrationQueryRuns) + .set({ + status: "succeeded", + result: safeResult, + responseMetadata: result.metadata, + completedAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq( + schema.integrationQueryRuns.organisationId, + input.actor.organisationId, + ), + eq(schema.integrationQueryRuns.id, run.id), + ), + ); + await tx + .update(schema.integrationRecords) + .set({ + status: "healthy", + health: { + status: "healthy", + checkedAt: new Date().toISOString(), + lastQueryRunId: run.id, + }, + lastSyncAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq( + schema.integrationRecords.organisationId, + input.actor.organisationId, + ), + eq(schema.integrationRecords.id, input.row.integration.id), + ), + ); + await appendAuditEvent(tx, { + organisationId: input.actor.organisationId, + actorId: input.actor.id, + actorType: input.actor.actorType, + action: "connector.query.succeeded", + targetType: "integration_query", + targetId: run.id, + metadata: { + source: "agent-live-context", + agentRunId: input.runId, + integrationId: input.row.integration.id, + templateKey: definition.key, + templateVersion: definition.version, + ...result.metadata, + }, + traceId: input.traceId, + }); + await writeOutbox(tx, { + organisationId: input.actor.organisationId, + eventType: "connector.query.succeeded", + aggregateType: "integration_query", + aggregateId: run.id, + queueName: "muster-outbox", + payload: { + queryRunId: run.id, + agentRunId: input.runId, + templateKey: definition.key, + }, + idempotencyKey: `connector.query.succeeded:${run.id}`, + traceId: input.traceId, + }); + }); + return { + queryRunId: run.id, + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: definition.key, + status: "succeeded", + result: safeResult, + responseMetadata: result.metadata, + }; + } catch (error) { + const failure = connectorFailure(error); + await input.db.transaction(async (tx) => { + await tx + .update(schema.integrationQueryRuns) + .set({ + status: "failed", + errorCode: failure.code, + errorMessage: failure.message, + completedAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq( + schema.integrationQueryRuns.organisationId, + input.actor.organisationId, + ), + eq(schema.integrationQueryRuns.id, run.id), + ), + ); + await appendAuditEvent(tx, { + organisationId: input.actor.organisationId, + actorId: input.actor.id, + actorType: input.actor.actorType, + action: "connector.query.failed", + targetType: "integration_query", + targetId: run.id, + metadata: { + source: "agent-live-context", + agentRunId: input.runId, + integrationId: input.row.integration.id, + templateKey: definition.key, + errorCode: failure.code, + }, + traceId: input.traceId, + }); + await writeOutbox(tx, { + organisationId: input.actor.organisationId, + eventType: "connector.query.failed", + aggregateType: "integration_query", + aggregateId: run.id, + queueName: "muster-outbox", + payload: { + queryRunId: run.id, + agentRunId: input.runId, + templateKey: definition.key, + errorCode: failure.code, + }, + idempotencyKey: `connector.query.failed:${run.id}`, + traceId: input.traceId, + }); + }); + return { + queryRunId: run.id, + source: input.row.integration.displayName, + product: input.row.integration.product, + templateKey: definition.key, + status: "failed", + errorCode: failure.code, + errorMessage: failure.message, + }; + } +} + +async function loadLiveConnectorEvidence(input: { + db: Db; + actor: typeof schema.actors.$inferSelect; + runId: string; + request: PersistedRequest; +}): Promise { + if ( + input.request.harness?.mode !== "slack" || + !input.request.humanRequest?.trim() + ) + return []; + const prompt = input.request.humanRequest.toLowerCase(); + const requestedProducts = new Set(); + if (/\b(tawny|host|hosts|endpoint|endpoints|machine|machines)\b/.test(prompt)) + requestedProducts.add("tawny"); + if (/\b(kelpie|case|cases|incident|incidents)\b/.test(prompt)) + requestedProducts.add("kelpie"); + if ( + /\b(unifi|network|traffic|client|clients|device|devices|bandwidth)\b/.test( + prompt, + ) + ) + requestedProducts.add("unifi"); + const products = requestedProducts.size + ? [...requestedProducts] + : ["tawny", "kelpie", "unifi"]; + const rows = await input.db + .select({ + integration: schema.integrationRecords, + template: schema.integrationQueryTemplates, + credential: schema.integrationConnectorCredentials, + }) + .from(schema.integrationRecords) + .innerJoin( + schema.integrationQueryTemplates, + and( + eq( + schema.integrationQueryTemplates.organisationId, + input.actor.organisationId, + ), + eq( + schema.integrationQueryTemplates.integrationId, + schema.integrationRecords.id, + ), + eq(schema.integrationQueryTemplates.enabled, true), + ), + ) + .innerJoin( + schema.integrationConnectorCredentials, + and( + eq( + schema.integrationConnectorCredentials.organisationId, + input.actor.organisationId, + ), + eq( + schema.integrationConnectorCredentials.integrationId, + schema.integrationRecords.id, + ), + ), + ) + .where( + and( + eq( + schema.integrationRecords.organisationId, + input.actor.organisationId, + ), + inArray(schema.integrationRecords.product, products), + inArray(schema.integrationRecords.status, ["configured", "healthy"]), + eq(schema.integrationRecords.mock, false), + isNull(schema.integrationRecords.archivedAt), + ), + ) + .orderBy(asc(schema.integrationRecords.createdAt)); + const evidence: LiveConnectorEvidence[] = []; + const traceId = redactObservationText( + input.request.traceId ?? `agent-run-${input.runId}`, + ); + const execute = async ( + product: string, + templateKey: string, + values: Record, + suffix?: string, + ) => { + const row = rows.find( + (candidate) => + candidate.integration.product === product && + candidate.template.templateKey === templateKey, + ); + if (!row) { + const unavailable: LiveConnectorEvidence = { + source: product, + product, + templateKey, + status: "unavailable", + errorCode: "integration_unavailable", + errorMessage: `No healthy ${product} connector template is configured.`, + }; + evidence.push(unavailable); + return unavailable; + } + const result = await executeLiveContextQuery({ + db: input.db, + row, + actor: input.actor, + runId: input.runId, + traceId, + values, + ...(suffix ? { suffix } : {}), + }); + evidence.push(result); + return result; + }; + await Promise.all([ + ...(requestedProducts.has("tawny") || requestedProducts.size === 0 + ? [execute("tawny", "tawny.inventory.list", {})] + : []), + ...(requestedProducts.has("kelpie") || requestedProducts.size === 0 + ? [execute("kelpie", "kelpie.cases.list", {})] + : []), + ]); + if (!requestedProducts.has("unifi") && requestedProducts.size !== 0) + return evidence; + const [sites] = await Promise.all([ + execute("unifi", "unifi.sites.list", { + offset: 0, + limit: 10, + }), + execute("unifi", "unifi.traffic.clients", { + siteName: "default", + }), + ]); + const siteIds = Array.isArray(sites.result) + ? sites.result + .map((site) => + site && typeof site === "object" + ? (site as Record).id + : undefined, + ) + .filter((siteId): siteId is string => typeof siteId === "string") + .slice(0, 3) + : []; + await Promise.all( + siteIds.map((siteId) => + execute( + "unifi", + "unifi.clients.list", + { siteId, offset: 0, limit: 50 }, + siteId, + ), + ), + ); + return evidence; +} + function promptParts( context: Context, request: PersistedRequest, @@ -1566,6 +2437,20 @@ function promptParts( }, ), })), + ...context.liveConnectorEvidence.map((query) => ({ + kind: "tool_result" as const, + tool: `${query.product}.${query.templateKey}`, + content: JSON.stringify({ + queryRunId: query.queryRunId, + source: query.source, + status: query.status, + result: query.result, + responseMetadata: query.responseMetadata, + errorCode: query.errorCode, + errorMessage: query.errorMessage, + trust: "untrusted-evidence", + }), + })), ]; } diff --git a/apps/agent-gateway/src/service-auth.test.ts b/apps/agent-gateway/src/service-auth.test.ts new file mode 100644 index 0000000..79721f2 --- /dev/null +++ b/apps/agent-gateway/src/service-auth.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from "vitest"; +import { + isGatewayRequestAuthorised, + parseGatewayOrganisationId, +} from "./service-auth.ts"; + +const token = "synthetic-agent-gateway-token-at-least-32-bytes"; + +describe("agent gateway service authentication", () => { + it("accepts only the exact bearer token", () => { + expect(isGatewayRequestAuthorised(`Bearer ${token}`, token)).toBe(true); + expect(isGatewayRequestAuthorised(undefined, token)).toBe(false); + expect(isGatewayRequestAuthorised(`Basic ${token}`, token)).toBe(false); + expect(isGatewayRequestAuthorised(`Bearer ${token}x`, token)).toBe(false); + expect( + isGatewayRequestAuthorised( + "Bearer synthetic-agent-gateway-token-wrong-value", + token, + ), + ).toBe(false); + }); + + it("accepts one valid organisation UUID header", () => { + const organisationId = "019fa127-8566-770b-939a-971ce03829f6"; + expect(parseGatewayOrganisationId(organisationId)).toBe(organisationId); + expect(parseGatewayOrganisationId(undefined)).toBeNull(); + expect(parseGatewayOrganisationId("not-an-id")).toBeNull(); + expect(parseGatewayOrganisationId([organisationId])).toBeNull(); + }); +}); diff --git a/apps/agent-gateway/src/service-auth.ts b/apps/agent-gateway/src/service-auth.ts new file mode 100644 index 0000000..7d8be26 --- /dev/null +++ b/apps/agent-gateway/src/service-auth.ts @@ -0,0 +1,21 @@ +import { timingSafeEqual } from "node:crypto"; +import { z } from "zod"; + +export function isGatewayRequestAuthorised( + authorization: string | undefined, + expectedToken: string, +) { + if (!authorization?.startsWith("Bearer ")) return false; + const supplied = Buffer.from(authorization.slice("Bearer ".length), "utf8"); + const expected = Buffer.from(expectedToken, "utf8"); + return ( + supplied.length === expected.length && timingSafeEqual(supplied, expected) + ); +} + +export function parseGatewayOrganisationId( + value: string | string[] | undefined, +) { + const parsed = z.string().uuid().safeParse(value); + return parsed.success ? parsed.data : null; +} diff --git a/apps/web/app/agent-runs/[id]/page.tsx b/apps/web/app/agent-runs/[id]/page.tsx new file mode 100644 index 0000000..dfacd6f --- /dev/null +++ b/apps/web/app/agent-runs/[id]/page.tsx @@ -0,0 +1,10 @@ +import { AgentRunView } from "@/components/agent-run-view"; + +export default async function AgentRunPage({ + params, +}: { + params: Promise<{ id: string }>; +}) { + const { id } = await params; + return ; +} diff --git a/apps/web/app/api/v1/agent-harness/invocations/route.ts b/apps/web/app/api/v1/agent-harness/invocations/route.ts new file mode 100644 index 0000000..edf048c --- /dev/null +++ b/apps/web/app/api/v1/agent-harness/invocations/route.ts @@ -0,0 +1,20 @@ +import { GovernedAgentHarness } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; + +export async function POST(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + const idempotencyKey = request.headers.get("idempotency-key")?.trim(); + if (!idempotencyKey) + throw new Error("Idempotency-Key header is required for harness invocations"); + const data = await new GovernedAgentHarness().invoke( + subject, + await request.json(), + idempotencyKey, + ); + return Response.json({ data, traceId }, { status: data.duplicate ? 200 : 202 }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/agent-harness/manifests/route.ts b/apps/web/app/api/v1/agent-harness/manifests/route.ts new file mode 100644 index 0000000..77f0723 --- /dev/null +++ b/apps/web/app/api/v1/agent-harness/manifests/route.ts @@ -0,0 +1,16 @@ +import { GovernedAgentHarness } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; + +export async function GET(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + return Response.json({ + protocolVersion: "muster.agent-harness/v1", + data: await new GovernedAgentHarness().manifest(subject), + traceId, + }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/agent-harness/runs/[id]/route.ts b/apps/web/app/api/v1/agent-harness/runs/[id]/route.ts new file mode 100644 index 0000000..595285e --- /dev/null +++ b/apps/web/app/api/v1/agent-harness/runs/[id]/route.ts @@ -0,0 +1,42 @@ +import { requireCapability } from "@muster/authz"; +import { GovernedAgentHarness } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; +import { agentGatewayHeaders } from "@/lib/agent-gateway"; + +type Context = { params: Promise<{ id: string }> }; + +export async function GET(request: Request, { params }: Context) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + const { id } = await params; + return Response.json({ + data: await new GovernedAgentHarness().read(subject, id), + traceId, + }); + } catch (error) { + return problemResponse(error, traceId); + } +} + +export async function DELETE(request: Request, { params }: Context) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + requireCapability(subject, "agents.cancel"); + const { id } = await params; + await new GovernedAgentHarness().read(subject, id); + const gateway = await fetch( + `${process.env.AGENT_GATEWAY_URL ?? "http://agent-gateway:3002"}/v1/runs/${encodeURIComponent(id)}/cancel`, + { + headers: agentGatewayHeaders(subject.organisationId), + method: "POST", + signal: AbortSignal.timeout(5_000), + }, + ); + if (!gateway.ok) throw new Error("Agent runtime did not accept cancellation"); + return Response.json({ data: await gateway.json(), traceId }, { status: 202 }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/agent-runs/[id]/route.ts b/apps/web/app/api/v1/agent-runs/[id]/route.ts index 75e8fb6..1a845c8 100644 --- a/apps/web/app/api/v1/agent-runs/[id]/route.ts +++ b/apps/web/app/api/v1/agent-runs/[id]/route.ts @@ -8,6 +8,7 @@ import { problemResponse, requestTraceId, } from "@/lib/api-context"; +import { agentGatewayHeaders } from "@/lib/agent-gateway"; import { settleAgentRun, type AgentRunResult } from "@/lib/task-domain"; export async function GET( @@ -35,7 +36,10 @@ export async function GET( } const gateway = await fetch( `${process.env.AGENT_GATEWAY_URL ?? "http://agent-gateway:3002"}/v1/runs/${encodeURIComponent(id)}`, - { signal: AbortSignal.timeout(5_000) }, + { + headers: agentGatewayHeaders(subject.organisationId), + signal: AbortSignal.timeout(5_000), + }, ); const result = (await gateway.json()) as { status?: string; diff --git a/apps/web/app/api/v1/agents/[id]/learning/route.ts b/apps/web/app/api/v1/agents/[id]/learning/route.ts index 4e428b6..9c44869 100644 --- a/apps/web/app/api/v1/agents/[id]/learning/route.ts +++ b/apps/web/app/api/v1/agents/[id]/learning/route.ts @@ -22,7 +22,12 @@ export async function GET( const subject = await apiSubject(request); requireCapability(subject, "agents.read"); const { id } = await params; - const data = await agentLearningState(subject.organisationId, id); + const includeInactive = + new URL(request.url).searchParams.get("includeInactive") === "true"; + if (includeInactive) requireCapability(subject, "agents.manage"); + const data = await agentLearningState(subject.organisationId, id, { + includeInactive, + }); return Response.json({ data, traceId }); } catch (error) { return problemResponse(error, traceId); diff --git a/apps/web/app/api/v1/hunts/route.ts b/apps/web/app/api/v1/hunts/route.ts index 01ba1ed..e454ad9 100644 --- a/apps/web/app/api/v1/hunts/route.ts +++ b/apps/web/app/api/v1/hunts/route.ts @@ -1,6 +1,6 @@ import { requireCapability } from "@muster/authz"; import { database, schema } from "@muster/database"; -import { desc, eq } from "drizzle-orm"; +import { and, desc, eq, isNull } from "drizzle-orm"; import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; import { JessieHuntDomainService } from "@/lib/jessie-hunt-domain"; @@ -12,7 +12,12 @@ export async function GET(request: Request) { const rows = await database() .select() .from(schema.huntRuns) - .where(eq(schema.huntRuns.organisationId, subject.organisationId)) + .where( + and( + eq(schema.huntRuns.organisationId, subject.organisationId), + isNull(schema.huntRuns.archivedAt), + ), + ) .orderBy(desc(schema.huntRuns.createdAt)) .limit(100); return Response.json({ data: rows, traceId }); diff --git a/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts b/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts new file mode 100644 index 0000000..9dbd574 --- /dev/null +++ b/apps/web/app/api/v1/maintenance/synthetic-cleanup/route.ts @@ -0,0 +1,18 @@ +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; +import { SyntheticCleanupDomainService } from "@/lib/synthetic-cleanup-domain"; + +export async function POST(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + const body = await request.json(); + const data = await new SyntheticCleanupDomainService().execute( + subject, + body, + traceId, + ); + return Response.json({ data, traceId }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/ready/route.test.ts b/apps/web/app/api/v1/ready/route.test.ts new file mode 100644 index 0000000..7057223 --- /dev/null +++ b/apps/web/app/api/v1/ready/route.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from "vitest"; +import { readinessResponse } from "./route.ts"; + +describe("GET /api/v1/ready", () => { + it("returns non-2xx when a required dependency is unavailable", async () => { + const response = readinessResponse({ + status: "degraded", + dependencies: [ + { name: "postgresql", status: "ready" }, + { name: "redis", status: "unavailable" }, + ], + }); + + expect(response.status).toBe(503); + await expect(response.json()).resolves.toEqual({ + status: "degraded", + dependencies: [ + { name: "postgresql", status: "ready" }, + { name: "redis", status: "unavailable" }, + ], + }); + }); +}); diff --git a/apps/web/app/api/v1/ready/route.ts b/apps/web/app/api/v1/ready/route.ts index 1a3649f..6b9278e 100644 --- a/apps/web/app/api/v1/ready/route.ts +++ b/apps/web/app/api/v1/ready/route.ts @@ -1 +1,16 @@ -export { GET } from "../health/route"; +import { + musterReadiness, + type ReadinessReport, +} from "../../../../lib/readiness.ts"; + +export const dynamic = "force-dynamic"; + +export function readinessResponse(report: ReadinessReport) { + return Response.json(report, { + status: report.status === "ready" ? 200 : 503, + }); +} + +export async function GET() { + return readinessResponse(await musterReadiness()); +} diff --git a/apps/web/app/api/v1/reports/route.ts b/apps/web/app/api/v1/reports/route.ts index fa0b83a..71a3ccd 100644 --- a/apps/web/app/api/v1/reports/route.ts +++ b/apps/web/app/api/v1/reports/route.ts @@ -1,6 +1,6 @@ import { requireCapability } from "@muster/authz"; import { database, schema } from "@muster/database"; -import { and, desc, eq } from "drizzle-orm"; +import { and, desc, eq, isNull } from "drizzle-orm"; import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; import { ParkerReportDomainService } from "@/lib/parker-report-domain"; @@ -9,25 +9,47 @@ export async function GET(request: Request) { try { const subject = await apiSubject(request); requireCapability(subject, "agents.read"); - const data = await database().select({ report: schema.reportManifests }).from(schema.reportManifests) + const data = await database() + .select({ report: schema.reportManifests }) + .from(schema.reportManifests) .innerJoin( schema.roomMemberships, and( - eq(schema.roomMemberships.organisationId, schema.reportManifests.organisationId), + eq( + schema.roomMemberships.organisationId, + schema.reportManifests.organisationId, + ), eq(schema.roomMemberships.roomId, schema.reportManifests.roomId), eq(schema.roomMemberships.actorId, subject.actorId), ), ) - .where(eq(schema.reportManifests.organisationId, subject.organisationId)) - .orderBy(desc(schema.reportManifests.createdAt)).limit(100); + .where( + and( + eq(schema.reportManifests.organisationId, subject.organisationId), + isNull(schema.reportManifests.archivedAt), + ), + ) + .orderBy(desc(schema.reportManifests.createdAt)) + .limit(100); return Response.json({ data: data.map((row) => row.report), traceId }); - } catch (error) { return problemResponse(error, traceId); } + } catch (error) { + return problemResponse(error, traceId); + } } export async function POST(request: Request) { const traceId = requestTraceId(request); try { - const result = await new ParkerReportDomainService().create(await apiSubject(request), await request.json(), traceId); - return Response.json({ data: result, traceId }, { status: result.duplicate ? 200 : 202 }); - } catch (error) { return problemResponse(error, traceId); } + const result = await new ParkerReportDomainService().create( + await apiSubject(request), + await request.json(), + traceId, + ); + return Response.json( + { data: result, traceId }, + { status: result.duplicate ? 200 : 202 }, + ); + } catch (error) { + return problemResponse(error, traceId); + } } diff --git a/apps/web/app/api/v1/reports/schedules/route.ts b/apps/web/app/api/v1/reports/schedules/route.ts index 05b50e8..93518eb 100644 --- a/apps/web/app/api/v1/reports/schedules/route.ts +++ b/apps/web/app/api/v1/reports/schedules/route.ts @@ -1,21 +1,42 @@ import { requireCapability } from "@muster/authz"; import { database, schema } from "@muster/database"; -import { and, desc, eq } from "drizzle-orm"; +import { and, desc, eq, isNull } from "drizzle-orm"; import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; import { ParkerReportDomainService } from "@/lib/parker-report-domain"; export async function GET(request: Request) { const traceId = requestTraceId(request); try { - const subject = await apiSubject(request); requireCapability(subject, "administration.manage"); - const data = await database().select().from(schema.reportSchedules).where(eq(schema.reportSchedules.organisationId, subject.organisationId)).orderBy(desc(schema.reportSchedules.nextRunAt)); + const subject = await apiSubject(request); + requireCapability(subject, "administration.manage"); + const data = await database() + .select() + .from(schema.reportSchedules) + .where( + and( + eq(schema.reportSchedules.organisationId, subject.organisationId), + isNull(schema.reportSchedules.archivedAt), + ), + ) + .orderBy(desc(schema.reportSchedules.nextRunAt)); return Response.json({ data, traceId }); - } catch (error) { return problemResponse(error, traceId); } + } catch (error) { + return problemResponse(error, traceId); + } } export async function POST(request: Request) { const traceId = requestTraceId(request); try { - const result = await new ParkerReportDomainService().createSchedule(await apiSubject(request), await request.json(), traceId); - return Response.json({ data: result, traceId }, { status: result.duplicate ? 200 : 201 }); - } catch (error) { return problemResponse(error, traceId); } + const result = await new ParkerReportDomainService().createSchedule( + await apiSubject(request), + await request.json(), + traceId, + ); + return Response.json( + { data: result, traceId }, + { status: result.duplicate ? 200 : 201 }, + ); + } catch (error) { + return problemResponse(error, traceId); + } } diff --git a/apps/web/app/api/v1/rooms/[id]/messages/route.ts b/apps/web/app/api/v1/rooms/[id]/messages/route.ts index 757e4c5..41b32e1 100644 --- a/apps/web/app/api/v1/rooms/[id]/messages/route.ts +++ b/apps/web/app/api/v1/rooms/[id]/messages/route.ts @@ -4,6 +4,7 @@ import { PostMessageSchema, RoomService } from "@muster/rooms"; import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; import { enforceApiRateLimit } from "@/lib/api-rate-limit"; import { publishRealtime } from "@/lib/realtime"; +import { AgentDirectMessageDomainService } from "@/lib/agent-direct-message-domain"; import { JessieHuntDomainService } from "@/lib/jessie-hunt-domain"; export async function GET( @@ -48,11 +49,28 @@ export async function POST( roomId: id, }); const result = await new RoomService().postMessage(subject, input, traceId); + let agentInvocation: Awaited< + ReturnType + > = null; + let agentInvocationError: string | null = null; let jessieHunt: Awaited< ReturnType > = null; let jessieHuntError: string | null = null; - if (result.created) { + try { + agentInvocation = await new AgentDirectMessageDomainService().maybeQueue( + subject, + { messageId: result.message.id, roomId: id }, + traceId, + ); + } catch (error) { + agentInvocationError = redactObservationText( + error instanceof Error + ? error.message + : "The direct-message agent could not be queued.", + ); + } + if (result.created && !agentInvocation && !agentInvocationError) { try { jessieHunt = await new JessieHuntDomainService().maybeCreateFromMention( subject, @@ -91,6 +109,8 @@ export async function POST( { data: result.message, duplicate: !result.created, + agentInvocation, + agentInvocationError, jessieHunt, jessieHuntError, realtimeDegraded: !realtimeDelivered, diff --git a/apps/web/app/api/v1/slack/commands/route.ts b/apps/web/app/api/v1/slack/commands/route.ts new file mode 100644 index 0000000..2222b9c --- /dev/null +++ b/apps/web/app/api/v1/slack/commands/route.ts @@ -0,0 +1,40 @@ +import { SlackGovernanceAdapter, verifySlackRequest } from "@muster/agent-harness"; + +export async function POST(request: Request) { + const rawBody = await request.text(); + const signingSecret = process.env.SLACK_SIGNING_SECRET; + if ( + !signingSecret || + !verifySlackRequest( + rawBody, + request.headers.get("x-slack-request-timestamp"), + request.headers.get("x-slack-signature"), + signingSecret, + ) + ) + return Response.json({ error: "invalid Slack signature" }, { status: 401 }); + const values = new URLSearchParams(rawBody); + const teamId = values.get("team_id"); + const userId = values.get("user_id"); + const channelId = values.get("channel_id"); + if (!teamId || !userId || !channelId) + return Response.json({ error: "invalid Slack command" }, { status: 400 }); + try { + await new SlackGovernanceAdapter().recordEvent(rawBody, { + type: "slash_command", + team_id: teamId, + event_id: values.get("trigger_id") ?? undefined, + event: { + type: "slash_command", + user: userId, + channel: channelId, + channel_type: values.get("channel_name") === "directmessage" ? "im" : "channel", + text: values.get("text") ?? "", + }, + }); + } catch { + // The signed request is acknowledged within Slack's deadline; outbox retry + // and installation health hold the durable failure path. + } + return Response.json({ response_type: "ephemeral", text: "Muster accepted your request." }); +} diff --git a/apps/web/app/api/v1/slack/events/route.ts b/apps/web/app/api/v1/slack/events/route.ts new file mode 100644 index 0000000..48e3f18 --- /dev/null +++ b/apps/web/app/api/v1/slack/events/route.ts @@ -0,0 +1,31 @@ +import { SlackGovernanceAdapter, verifySlackRequest } from "@muster/agent-harness"; + +export async function POST(request: Request) { + const rawBody = await request.text(); + const signingSecret = process.env.SLACK_SIGNING_SECRET; + if ( + !signingSecret || + !verifySlackRequest( + rawBody, + request.headers.get("x-slack-request-timestamp"), + request.headers.get("x-slack-signature"), + signingSecret, + ) + ) + return Response.json({ error: "invalid Slack signature" }, { status: 401 }); + let payload: Record; + try { + payload = JSON.parse(rawBody) as Record; + } catch { + return Response.json({ error: "invalid Slack payload" }, { status: 400 }); + } + if (payload.type === "url_verification" && typeof payload.challenge === "string") + return Response.json({ challenge: payload.challenge }); + try { + await new SlackGovernanceAdapter().recordEvent(rawBody, payload); + } catch { + // Acknowledge retried or currently unmapped events. Installation health + // records retain the authoritative failure path without leaking it to Slack. + } + return Response.json({ ok: true }); +} diff --git a/apps/web/app/api/v1/slack/exposures/route.ts b/apps/web/app/api/v1/slack/exposures/route.ts new file mode 100644 index 0000000..3d3a4ec --- /dev/null +++ b/apps/web/app/api/v1/slack/exposures/route.ts @@ -0,0 +1,39 @@ +import { SlackGovernanceAdapter } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; +import { z } from "zod"; + +const ExposureSchema = z.object({ + installationId: z.string().uuid(), + agentId: z.string().uuid(), + enabled: z.boolean(), + isDefault: z.boolean(), + allowedChannelIds: z.array(z.string().trim().min(1).max(128)).max(500).optional(), + allowDirectMessages: z.boolean().optional(), + allowThreadContext: z.boolean().optional(), +}); + +export async function PUT(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + const input = ExposureSchema.parse(await request.json()); + await new SlackGovernanceAdapter().configureExposure(subject, { + installationId: input.installationId, + agentId: input.agentId, + enabled: input.enabled, + isDefault: input.isDefault, + ...(input.allowedChannelIds === undefined + ? {} + : { allowedChannelIds: input.allowedChannelIds }), + ...(input.allowDirectMessages === undefined + ? {} + : { allowDirectMessages: input.allowDirectMessages }), + ...(input.allowThreadContext === undefined + ? {} + : { allowThreadContext: input.allowThreadContext }), + }); + return Response.json({ data: { status: "configured" }, traceId }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/slack/health/route.ts b/apps/web/app/api/v1/slack/health/route.ts new file mode 100644 index 0000000..683b2ff --- /dev/null +++ b/apps/web/app/api/v1/slack/health/route.ts @@ -0,0 +1,12 @@ +import { SlackGovernanceAdapter } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; + +export async function GET(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + return Response.json({ data: await new SlackGovernanceAdapter().health(subject), traceId }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/slack/identities/route.ts b/apps/web/app/api/v1/slack/identities/route.ts new file mode 100644 index 0000000..0001667 --- /dev/null +++ b/apps/web/app/api/v1/slack/identities/route.ts @@ -0,0 +1,23 @@ +import { SlackGovernanceAdapter } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; +import { z } from "zod"; + +const MappingSchema = z.object({ + installationId: z.string().uuid(), + slackUserId: z.string().trim().min(1).max(128), + actorId: z.string().uuid(), +}); + +export async function POST(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + await new SlackGovernanceAdapter().mapIdentity( + subject, + MappingSchema.parse(await request.json()), + ); + return Response.json({ data: { status: "active" }, traceId }, { status: 201 }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/slack/ingress.test.ts b/apps/web/app/api/v1/slack/ingress.test.ts new file mode 100644 index 0000000..afd0dca --- /dev/null +++ b/apps/web/app/api/v1/slack/ingress.test.ts @@ -0,0 +1,100 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const harness = vi.hoisted(() => ({ + recordEvent: vi.fn(), + verifySlackRequest: vi.fn(), +})); + +vi.mock("@muster/agent-harness", () => ({ + verifySlackRequest: harness.verifySlackRequest, + SlackGovernanceAdapter: class { + recordEvent = harness.recordEvent; + }, +})); + +import { POST as command } from "./commands/route.ts"; +import { POST as event } from "./events/route.ts"; +import { POST as interaction } from "./interactions/route.ts"; + +function request(body: string, headers: HeadersInit = {}) { + return new Request("https://muster.example/api/v1/slack/test", { + method: "POST", + headers: { + "content-type": "application/json", + "x-slack-request-timestamp": "1700000000", + "x-slack-signature": "v0=synthetic", + ...headers, + }, + body, + }); +} + +describe("Slack signed ingress", () => { + const originalSecret = process.env.SLACK_SIGNING_SECRET; + + beforeEach(() => { + process.env.SLACK_SIGNING_SECRET = "synthetic-signing-secret"; + harness.recordEvent.mockReset(); + harness.verifySlackRequest.mockReset().mockReturnValue(true); + }); + + afterEach(() => { + if (originalSecret === undefined) delete process.env.SLACK_SIGNING_SECRET; + else process.env.SLACK_SIGNING_SECRET = originalSecret; + }); + + it("rejects unsigned Events API payloads before parsing or persistence", async () => { + harness.verifySlackRequest.mockReturnValue(false); + const response = await event(request('{"type":"event_callback"}')); + expect(response.status).toBe(401); + expect(harness.recordEvent).not.toHaveBeenCalled(); + }); + + it("answers only signed Slack URL verification without creating an inbox event", async () => { + const response = await event( + request('{"type":"url_verification","challenge":"synthetic-challenge"}'), + ); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ + challenge: "synthetic-challenge", + }); + expect(harness.recordEvent).not.toHaveBeenCalled(); + }); + + it("normalises a signed slash command into durable, direct-message ingress", async () => { + const body = new URLSearchParams({ + team_id: "T-synthetic", + user_id: "U-synthetic", + channel_id: "D-synthetic", + channel_name: "directmessage", + trigger_id: "trigger-synthetic", + text: "Jessie bounded synthetic request", + }).toString(); + const response = await command( + request(body, { "content-type": "application/x-www-form-urlencoded" }), + ); + expect(response.status).toBe(200); + expect(harness.recordEvent).toHaveBeenCalledWith(body, { + type: "slash_command", + team_id: "T-synthetic", + event_id: "trigger-synthetic", + event: { + type: "slash_command", + user: "U-synthetic", + channel: "D-synthetic", + channel_type: "im", + text: "Jessie bounded synthetic request", + }, + }); + }); + + it("rejects malformed signed interaction bodies without invoking an adapter", async () => { + const response = await interaction( + request("not-a-slack-interaction", { + "content-type": "application/x-www-form-urlencoded", + }), + ); + expect(response.status).toBe(400); + expect(harness.recordEvent).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/web/app/api/v1/slack/install/route.ts b/apps/web/app/api/v1/slack/install/route.ts new file mode 100644 index 0000000..eb69da8 --- /dev/null +++ b/apps/web/app/api/v1/slack/install/route.ts @@ -0,0 +1,45 @@ +import { requireCapability } from "@muster/authz"; +import { + requiredSlackBotScopes, + SlackGovernanceAdapter, + signSlackOAuthState, +} from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; + +export async function GET(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + requireCapability(subject, "administration.manage"); + const clientId = process.env.SLACK_CLIENT_ID; + const redirectUri = process.env.SLACK_REDIRECT_URI; + if (!clientId || !redirectUri) throw new Error("Slack OAuth is not configured"); + const state = signSlackOAuthState({ + organisationId: subject.organisationId, + actorId: subject.actorId, + expiresAt: Date.now() + 10 * 60_000, + }); + const authorizationUrl = new URL("https://slack.com/oauth/v2/authorize"); + authorizationUrl.searchParams.set("client_id", clientId); + authorizationUrl.searchParams.set("redirect_uri", redirectUri); + authorizationUrl.searchParams.set("scope", requiredSlackBotScopes.join(",")); + authorizationUrl.searchParams.set("state", state); + return Response.json({ data: { authorizationUrl: authorizationUrl.toString() }, traceId }); + } catch (error) { + return problemResponse(error, traceId); + } +} + +export async function DELETE(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + requireCapability(subject, "administration.manage"); + const installationId = new URL(request.url).searchParams.get("installationId"); + if (!installationId) throw new Error("Slack installation id is required"); + await new SlackGovernanceAdapter().revoke(subject, installationId); + return Response.json({ data: { status: "revoked" }, traceId }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/slack/interactions/route.ts b/apps/web/app/api/v1/slack/interactions/route.ts new file mode 100644 index 0000000..7b76128 --- /dev/null +++ b/apps/web/app/api/v1/slack/interactions/route.ts @@ -0,0 +1,26 @@ +import { SlackGovernanceAdapter, verifySlackRequest } from "@muster/agent-harness"; + +export async function POST(request: Request) { + const rawBody = await request.text(); + const signingSecret = process.env.SLACK_SIGNING_SECRET; + if ( + !signingSecret || + !verifySlackRequest( + rawBody, + request.headers.get("x-slack-request-timestamp"), + request.headers.get("x-slack-signature"), + signingSecret, + ) + ) + return Response.json({ error: "invalid Slack signature" }, { status: 401 }); + const encoded = new URLSearchParams(rawBody).get("payload"); + if (!encoded) return Response.json({ error: "invalid Slack interaction" }, { status: 400 }); + try { + const payload = JSON.parse(encoded) as Record; + await new SlackGovernanceAdapter().recordEvent(rawBody, payload); + } catch { + // Slack requires a fast acknowledgement. The durable inbox carries retries + // and operator-visible delivery health without exposing internals to Slack. + } + return Response.json({ ok: true }); +} diff --git a/apps/web/app/api/v1/slack/lifecycle.integration.test.ts b/apps/web/app/api/v1/slack/lifecycle.integration.test.ts new file mode 100644 index 0000000..a9fac32 --- /dev/null +++ b/apps/web/app/api/v1/slack/lifecycle.integration.test.ts @@ -0,0 +1,694 @@ +import { createHash, createHmac } from "node:crypto"; +import { createServer, type Server } from "node:http"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + decryptConnectorPayload, + processSlackNotificationJob, + requiredSlackBotScopes, + SlackGovernanceAdapter, +} from "@muster/agent-harness"; +import { FakeSlackServer } from "@muster/agent-harness/testing/fake-slack-server"; +import { + closeDatabase, + database, + markOutboxDispatched, + schema, + writeOutbox, +} from "@muster/database"; +import { and, eq, inArray, like } from "drizzle-orm"; +import { POST as commandRoute } from "./commands/route"; +import { POST as eventRoute } from "./events/route"; +import { POST as interactionRoute } from "./interactions/route"; + +const integration = process.env.MUSTER_INTEGRATION_TESTS === "true"; +const describeIntegration = integration ? describe.sequential : describe.skip; + +class SlackIngressServer { + private server: Server | undefined; + private origin: string | undefined; + + async start() { + this.server = createServer(async (request, response) => { + const body = await new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => + resolve(Buffer.concat(chunks).toString("utf8")), + ); + request.on("error", reject); + }); + if ( + request.url !== "/api/v1/slack/events" && + request.url !== "/api/v1/slack/commands" && + request.url !== "/api/v1/slack/interactions" + ) { + response.writeHead(404).end(); + return; + } + const headers = new Headers(); + for (const [name, value] of Object.entries(request.headers)) { + if (Array.isArray(value)) { + for (const item of value) headers.append(name, item); + } else if (value !== undefined) { + headers.set(name, value); + } + } + const routedRequest = new Request( + `http://muster.synthetic${request.url}`, + { + method: "POST", + headers, + body, + }, + ); + const routed = + request.url === "/api/v1/slack/events" + ? await eventRoute(routedRequest) + : request.url === "/api/v1/slack/commands" + ? await commandRoute(routedRequest) + : await interactionRoute(routedRequest); + response.statusCode = routed.status; + routed.headers.forEach((value, name) => response.setHeader(name, value)); + response.end(await routed.text()); + }); + await new Promise((resolve, reject) => { + this.server!.once("error", reject); + this.server!.listen(0, "127.0.0.1", () => resolve()); + }); + const address = this.server.address(); + if (!address || typeof address === "string") + throw new Error("Synthetic Muster ingress did not bind an HTTP port"); + this.origin = `http://127.0.0.1:${address.port}`; + } + + async stop() { + const server = this.server; + this.server = undefined; + this.origin = undefined; + if (!server) return; + await new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ); + } + + async signedPost(path: string, body: string, contentType: string) { + if (!this.origin) + throw new Error("Synthetic Muster ingress is not running"); + const timestamp = String(Math.floor(Date.now() / 1_000)); + const signature = `v0=${createHmac( + "sha256", + process.env.SLACK_SIGNING_SECRET!, + ) + .update(`v0:${timestamp}:${body}`) + .digest("hex")}`; + return fetch(`${this.origin}${path}`, { + method: "POST", + headers: { + "content-type": contentType, + "x-slack-request-timestamp": timestamp, + "x-slack-signature": signature, + }, + body, + }); + } +} + +describeIntegration("hermetic Slack HTTP lifecycle", () => { + const db = database(); + const suffix = crypto.randomUUID(); + const teamId = `T-e2e-${suffix}`; + const slackUserId = `U-e2e-${suffix}`; + const botUserId = `B-e2e-${suffix}`; + const channelId = `C-e2e-${suffix}`; + const fakeSlack = new FakeSlackServer({ + teamId, + teamName: "Synthetic E2E Slack", + botUserId, + requiredScopes: requiredSlackBotScopes, + }); + const ingress = new SlackIngressServer(); + const originalEnvironment = { + clientId: process.env.SLACK_CLIENT_ID, + clientSecret: process.env.SLACK_CLIENT_SECRET, + signingSecret: process.env.SLACK_SIGNING_SECRET, + connectorKey: process.env.CONNECTOR_ENCRYPTION_KEY, + apiBaseUrl: process.env.MUSTER_TEST_SLACK_API_BASE_URL, + }; + let organisationId = ""; + let actorId = ""; + let agentId = ""; + let agentName = ""; + let installationId = ""; + + beforeAll(async () => { + process.env.SLACK_CLIENT_ID = "synthetic-client"; + process.env.SLACK_CLIENT_SECRET = "synthetic-client-secret"; + process.env.SLACK_SIGNING_SECRET = "synthetic-signing-secret"; + process.env.CONNECTOR_ENCRYPTION_KEY = Buffer.alloc(32, 31).toString( + "base64", + ); + process.env.MUSTER_TEST_SLACK_API_BASE_URL = await fakeSlack.start(); + await ingress.start(); + + const actors = await db + .select({ + id: schema.actors.id, + organisationId: schema.actors.organisationId, + capabilities: schema.actors.capabilityAssignments, + }) + .from(schema.actors) + .where(eq(schema.actors.actorType, "human")); + const actor = actors.find( + (candidate) => + Array.isArray(candidate.capabilities) && + candidate.capabilities.includes("administration.manage") && + candidate.capabilities.includes("agents.invoke"), + ); + if (!actor) + throw new Error("Bootstrap a synthetic Muster administrator first"); + actorId = actor.id; + organisationId = actor.organisationId; + const [agent] = await db + .select({ + id: schema.agentDefinitions.id, + name: schema.agentDefinitions.name, + }) + .from(schema.agentDefinitions) + .where( + and( + eq(schema.agentDefinitions.organisationId, organisationId), + eq(schema.agentDefinitions.status, "active"), + eq(schema.agentDefinitions.killSwitch, false), + ), + ) + .limit(1); + if (!agent) throw new Error("Bootstrap a synthetic active agent first"); + agentId = agent.id; + agentName = agent.name; + }); + + afterAll(async () => { + await ingress.stop(); + await fakeSlack.stop(); + if (agentId) + await db + .update(schema.agentDefinitions) + .set({ killSwitch: false }) + .where(eq(schema.agentDefinitions.id, agentId)); + if (installationId) { + const runs = await db + .select({ id: schema.agentRuns.id }) + .from(schema.agentRuns) + .where( + like(schema.agentRuns.idempotencyKey, `slack:${installationId}:%`), + ); + const runIds = runs.map((run) => run.id); + const inbox = await db + .select({ id: schema.slackInboxEvents.id }) + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.installationId, installationId)); + const aggregateIds = [...runIds, ...inbox.map((event) => event.id)]; + if (aggregateIds.length) + await db + .delete(schema.outboxEvents) + .where(inArray(schema.outboxEvents.aggregateId, aggregateIds)); + await db + .delete(schema.slackRunDeliveries) + .where(eq(schema.slackRunDeliveries.installationId, installationId)); + await db + .delete(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.installationId, installationId)); + if (runIds.length) { + await db + .delete(schema.agentRunEvents) + .where(inArray(schema.agentRunEvents.runId, runIds)); + await db + .delete(schema.agentRuns) + .where(inArray(schema.agentRuns.id, runIds)); + } + await db + .delete(schema.slackAgentExposures) + .where(eq(schema.slackAgentExposures.installationId, installationId)); + await db + .delete(schema.slackIdentityMappings) + .where(eq(schema.slackIdentityMappings.installationId, installationId)); + await db + .delete(schema.slackInstallations) + .where(eq(schema.slackInstallations.id, installationId)); + } + await closeDatabase(); + for (const [name, value] of Object.entries({ + SLACK_CLIENT_ID: originalEnvironment.clientId, + SLACK_CLIENT_SECRET: originalEnvironment.clientSecret, + SLACK_SIGNING_SECRET: originalEnvironment.signingSecret, + CONNECTOR_ENCRYPTION_KEY: originalEnvironment.connectorKey, + MUSTER_TEST_SLACK_API_BASE_URL: originalEnvironment.apiBaseUrl, + })) { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; + } + }); + + it("runs OAuth, signed ingress, worker delivery, replay, throttling, and revocation over HTTP", async () => { + const adapter = new SlackGovernanceAdapter(db); + const administrator = { + actorId, + organisationId, + capabilities: new Set([ + "administration.manage", + "agents.invoke", + ] as const), + }; + await expect( + adapter.install( + administrator, + "missing-scopes", + "http://muster.synthetic/api/v1/slack/oauth/callback", + ), + ).rejects.toThrow("missing required bot scopes: commands"); + expect(fakeSlack.requestsFor("oauth.v2.access")).toHaveLength(1); + + const installation = await adapter.install( + administrator, + "valid-install", + "http://muster.synthetic/api/v1/slack/oauth/callback", + ); + installationId = installation.id; + expect(installation.scopes).toEqual( + expect.arrayContaining([...requiredSlackBotScopes]), + ); + await adapter.mapIdentity(administrator, { + installationId, + slackUserId, + actorId, + }); + await adapter.configureExposure(administrator, { + installationId, + agentId, + enabled: true, + isDefault: true, + allowedChannelIds: [channelId], + }); + + const verificationBody = JSON.stringify({ + type: "url_verification", + challenge: "synthetic-challenge", + }); + const verification = await ingress.signedPost( + "/api/v1/slack/events", + verificationBody, + "application/json", + ); + await expect(verification.json()).resolves.toEqual({ + challenge: "synthetic-challenge", + }); + + const mentionEventId = `Ev-mention-${suffix}`; + const mentionBody = JSON.stringify({ + type: "event_callback", + team_id: teamId, + event_id: mentionEventId, + event: { + type: "app_mention", + user: slackUserId, + channel: channelId, + ts: "1710000000.000100", + text: "Run the bounded synthetic review", + }, + }); + expect( + ( + await ingress.signedPost( + "/api/v1/slack/events", + mentionBody, + "application/json", + ) + ).status, + ).toBe(200); + await ingress.stop(); + await ingress.start(); + expect( + ( + await ingress.signedPost( + "/api/v1/slack/events", + mentionBody, + "application/json", + ) + ).status, + ).toBe(200); + + const mentionInboxes = await db + .select() + .from(schema.slackInboxEvents) + .where( + and( + eq(schema.slackInboxEvents.installationId, installationId), + eq(schema.slackInboxEvents.eventId, mentionEventId), + ), + ); + expect(mentionInboxes).toHaveLength(1); + const mentionInbox = mentionInboxes[0]!; + const mentionOutbox = await db + .select() + .from(schema.outboxEvents) + .where(eq(schema.outboxEvents.aggregateId, mentionInbox.id)); + expect(mentionOutbox).toHaveLength(1); + expect(mentionOutbox[0]?.eventType).toBe("slack.event.received"); + await markOutboxDispatched(db, mentionOutbox[0]!.id); + expect( + await processSlackNotificationJob( + "slack.event.received", + mentionInbox.id, + ), + ).toBe(true); + expect(fakeSlack.requestsFor("chat.postMessage")).toHaveLength(1); + + const slashTriggerId = `trigger-${suffix}`; + const slashBody = new URLSearchParams({ + team_id: teamId, + user_id: slackUserId, + channel_id: channelId, + channel_name: "synthetic", + trigger_id: slashTriggerId, + text: agentName, + }).toString(); + expect( + ( + await ingress.signedPost( + "/api/v1/slack/commands", + slashBody, + "application/x-www-form-urlencoded", + ) + ).status, + ).toBe(200); + const [slashInbox] = await db + .select() + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.eventId, slashTriggerId)); + await processSlackNotificationJob("slack.event.received", slashInbox!.id); + const [processedSlash] = await db + .select({ + status: schema.slackInboxEvents.status, + error: schema.slackInboxEvents.error, + }) + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.id, slashInbox!.id)); + expect(processedSlash).toEqual({ status: "processed", error: null }); + + const shortcutPayload = { + type: "message_action", + callback_id: "muster.review", + team: { id: teamId }, + user: { id: slackUserId }, + channel: { id: channelId }, + message: { + ts: "1710000001.000100", + text: "Treat this synthetic message as bounded evidence only", + }, + }; + const shortcutBody = new URLSearchParams({ + payload: JSON.stringify(shortcutPayload), + }).toString(); + expect( + ( + await ingress.signedPost( + "/api/v1/slack/interactions", + shortcutBody, + "application/x-www-form-urlencoded", + ) + ).status, + ).toBe(200); + const shortcutEventId = createHash("sha256") + .update(shortcutBody) + .digest("hex"); + const [shortcutInbox] = await db + .select() + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.eventId, shortcutEventId)); + await processSlackNotificationJob( + "slack.event.received", + shortcutInbox!.id, + ); + const [processedShortcut] = await db + .select({ + status: schema.slackInboxEvents.status, + error: schema.slackInboxEvents.error, + }) + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.id, shortcutInbox!.id)); + expect(processedShortcut).toEqual({ status: "processed", error: null }); + expect(fakeSlack.requestsFor("chat.postMessage")).toHaveLength(3); + + const mentionRuns = await db + .select() + .from(schema.agentRuns) + .where( + eq( + schema.agentRuns.idempotencyKey, + `slack:${installationId}:${mentionEventId}`, + ), + ); + expect(mentionRuns).toHaveLength(1); + const mentionRun = mentionRuns[0]!; + const firstCiphertext = installation.encryptedBotToken; + const rotated = await adapter.install( + administrator, + "rotate-install", + "http://muster.synthetic/api/v1/slack/oauth/callback", + ); + expect(rotated.id).toBe(installationId); + expect(rotated.encryptedBotToken).not.toBe(firstCiphertext); + const rotatedToken = ( + decryptConnectorPayload( + rotated.encryptedBotToken, + process.env.CONNECTOR_ENCRYPTION_KEY!, + ) as { token: string } + ).token; + + await db.transaction(async (tx) => { + await tx + .update(schema.agentRuns) + .set({ + status: "completed", + progress: { stage: "completed", percent: 100 }, + structuredOutput: { + summary: "Hermetic Slack lifecycle completed", + confidence: 1, + gaps: [], + }, + }) + .where(eq(schema.agentRuns.id, mentionRun.id)); + await writeOutbox(tx, { + organisationId, + eventType: "agent.run.settled", + aggregateType: "agent_run", + aggregateId: mentionRun.id, + queueName: "muster-notifications", + payload: { runId: mentionRun.id }, + idempotencyKey: `synthetic.slack.settled:${mentionRun.id}`, + traceId: mentionEventId, + }); + }); + const settledOutbox = await db + .select() + .from(schema.outboxEvents) + .where( + eq( + schema.outboxEvents.idempotencyKey, + `synthetic.slack.settled:${mentionRun.id}`, + ), + ); + expect(settledOutbox).toHaveLength(1); + await markOutboxDispatched(db, settledOutbox[0]!.id); + fakeSlack.rateLimitOnce("chat.update"); + await processSlackNotificationJob("agent.run.settled", mentionRun.id); + expect(fakeSlack.requestsFor("chat.update")).toHaveLength(2); + expect(fakeSlack.requestsFor("chat.update").at(-1)?.authorization).toBe( + `Bearer ${rotatedToken}`, + ); + const [deliveredMention] = await db + .select({ + status: schema.slackRunDeliveries.status, + attemptCount: schema.slackRunDeliveries.attemptCount, + }) + .from(schema.slackRunDeliveries) + .where(eq(schema.slackRunDeliveries.runId, mentionRun.id)); + expect(deliveredMention).toEqual({ + status: "delivered", + attemptCount: 0, + }); + + const callsBeforeKillSwitch = fakeSlack.requests.length; + await db + .update(schema.agentDefinitions) + .set({ killSwitch: true }) + .where(eq(schema.agentDefinitions.id, agentId)); + const killedEventId = `Ev-killed-${suffix}`; + const killedBody = JSON.stringify({ + type: "event_callback", + team_id: teamId, + event_id: killedEventId, + event: { + type: "app_mention", + user: slackUserId, + channel: channelId, + text: "This must remain blocked", + }, + }); + await ingress.signedPost( + "/api/v1/slack/events", + killedBody, + "application/json", + ); + const [killedInbox] = await db + .select() + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.eventId, killedEventId)); + await processSlackNotificationJob("slack.event.received", killedInbox!.id); + expect(fakeSlack.requests).toHaveLength(callsBeforeKillSwitch); + const [processedKilled] = await db + .select({ + status: schema.slackInboxEvents.status, + error: schema.slackInboxEvents.error, + }) + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.id, killedInbox!.id)); + expect(processedKilled).toEqual({ + status: "ignored", + error: "agent_not_exposed", + }); + await db + .update(schema.agentDefinitions) + .set({ killSwitch: false }) + .where(eq(schema.agentDefinitions.id, agentId)); + + const revokedBody = JSON.stringify({ + type: "event_callback", + team_id: teamId, + event_id: `Ev-tokens-revoked-${suffix}`, + event: { + type: "tokens_revoked", + tokens: { oauth: [], bot: [botUserId] }, + }, + }); + await ingress.signedPost( + "/api/v1/slack/events", + revokedBody, + "application/json", + ); + const [revokedInbox] = await db + .select() + .from(schema.slackInboxEvents) + .where( + eq(schema.slackInboxEvents.eventId, `Ev-tokens-revoked-${suffix}`), + ); + await processSlackNotificationJob("slack.event.received", revokedInbox!.id); + const callsBeforeBlockedDelivery = fakeSlack.requests.length; + const [shortcutRun] = await db + .select() + .from(schema.agentRuns) + .where( + eq( + schema.agentRuns.idempotencyKey, + `slack:${installationId}:${shortcutEventId}`, + ), + ); + await db + .update(schema.agentRuns) + .set({ status: "completed" }) + .where(eq(schema.agentRuns.id, shortcutRun!.id)); + await processSlackNotificationJob("agent.run.settled", shortcutRun!.id); + expect(fakeSlack.requests).toHaveLength(callsBeforeBlockedDelivery); + const [blockedShortcutDelivery] = await db + .select({ status: schema.slackRunDeliveries.status }) + .from(schema.slackRunDeliveries) + .where(eq(schema.slackRunDeliveries.runId, shortcutRun!.id)); + expect(blockedShortcutDelivery?.status).toBe("blocked"); + + await adapter.install( + administrator, + "reinstall-after-revoke", + "http://muster.synthetic/api/v1/slack/oauth/callback", + ); + const uninstallBody = JSON.stringify({ + type: "event_callback", + team_id: teamId, + event_id: `Ev-app-uninstalled-${suffix}`, + event: { type: "app_uninstalled" }, + }); + await ingress.signedPost( + "/api/v1/slack/events", + uninstallBody, + "application/json", + ); + const [uninstallInbox] = await db + .select() + .from(schema.slackInboxEvents) + .where( + eq(schema.slackInboxEvents.eventId, `Ev-app-uninstalled-${suffix}`), + ); + await processSlackNotificationJob( + "slack.event.received", + uninstallInbox!.id, + ); + const [finalInstallation] = await db + .select({ + status: schema.slackInstallations.status, + encryptedBotToken: schema.slackInstallations.encryptedBotToken, + }) + .from(schema.slackInstallations) + .where(eq(schema.slackInstallations.id, installationId)); + expect(finalInstallation?.status).toBe("revoked"); + expect( + decryptConnectorPayload( + finalInstallation!.encryptedBotToken, + process.env.CONNECTOR_ENCRYPTION_KEY!, + ), + ).toEqual({ revoked: true }); + const [slashRun] = await db + .select() + .from(schema.agentRuns) + .where( + eq( + schema.agentRuns.idempotencyKey, + `slack:${installationId}:${slashTriggerId}`, + ), + ); + const callsBeforeUninstalledDelivery = fakeSlack.requests.length; + await db + .update(schema.agentRuns) + .set({ status: "completed" }) + .where(eq(schema.agentRuns.id, slashRun!.id)); + await processSlackNotificationJob("agent.run.settled", slashRun!.id); + expect(fakeSlack.requests).toHaveLength(callsBeforeUninstalledDelivery); + const [blockedSlashDelivery] = await db + .select({ status: schema.slackRunDeliveries.status }) + .from(schema.slackRunDeliveries) + .where(eq(schema.slackRunDeliveries.runId, slashRun!.id)); + expect(blockedSlashDelivery?.status).toBe("blocked"); + const inboxCountBefore = await db + .select({ id: schema.slackInboxEvents.id }) + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.installationId, installationId)); + await ingress.signedPost( + "/api/v1/slack/events", + JSON.stringify({ + type: "event_callback", + team_id: teamId, + event_id: `Ev-after-uninstall-${suffix}`, + event: { + type: "app_mention", + user: slackUserId, + channel: channelId, + text: "Must not enter the durable inbox", + }, + }), + "application/json", + ); + const inboxCountAfter = await db + .select({ id: schema.slackInboxEvents.id }) + .from(schema.slackInboxEvents) + .where(eq(schema.slackInboxEvents.installationId, installationId)); + expect(inboxCountAfter).toHaveLength(inboxCountBefore.length); + }, 30_000); +}); diff --git a/apps/web/app/api/v1/slack/oauth/callback/route.ts b/apps/web/app/api/v1/slack/oauth/callback/route.ts new file mode 100644 index 0000000..9d78121 --- /dev/null +++ b/apps/web/app/api/v1/slack/oauth/callback/route.ts @@ -0,0 +1,61 @@ +import { + SlackGovernanceAdapter, + verifySlackOAuthState, +} from "@muster/agent-harness"; +import { capabilities, type AuthorisationSubject } from "@muster/authz"; +import { database, schema } from "@muster/database"; +import { and, eq } from "drizzle-orm"; +import { problemResponse, requestTraceId } from "@/lib/api-context"; + +export async function GET(request: Request) { + const traceId = requestTraceId(request); + try { + const url = new URL(request.url); + const code = url.searchParams.get("code"); + const state = url.searchParams.get("state"); + if (!code || !state) throw new Error("Slack OAuth callback is incomplete"); + const verified = verifySlackOAuthState(state); + const [actor] = await database() + .select({ capabilities: schema.actors.capabilityAssignments }) + .from(schema.actors) + .where( + and( + eq(schema.actors.id, verified.actorId), + eq(schema.actors.organisationId, verified.organisationId), + eq(schema.actors.actorType, "human"), + eq(schema.actors.status, "active"), + ), + ) + .limit(1); + if (!actor || !Array.isArray(actor.capabilities)) + throw new Error("Slack OAuth installer is no longer authorised"); + const subject: AuthorisationSubject = { + actorId: verified.actorId, + organisationId: verified.organisationId, + capabilities: new Set( + actor.capabilities.filter( + (capability): capability is (typeof capabilities)[number] => + typeof capability === "string" && + capabilities.includes(capability as (typeof capabilities)[number]), + ), + ), + }; + const slack = new SlackGovernanceAdapter(); + await slack.consumeOAuthState(subject, state); + const installation = await slack.install( + subject, + code, + process.env.SLACK_REDIRECT_URI ?? "", + ); + return Response.json({ + data: { + id: installation.id, + teamId: installation.teamId, + status: installation.status, + }, + traceId, + }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/slack/settings/route.ts b/apps/web/app/api/v1/slack/settings/route.ts new file mode 100644 index 0000000..edcc160 --- /dev/null +++ b/apps/web/app/api/v1/slack/settings/route.ts @@ -0,0 +1,15 @@ +import { SlackGovernanceAdapter } from "@muster/agent-harness"; +import { apiSubject, problemResponse, requestTraceId } from "@/lib/api-context"; + +export async function GET(request: Request) { + const traceId = requestTraceId(request); + try { + const subject = await apiSubject(request); + return Response.json({ + data: await new SlackGovernanceAdapter().settings(subject), + traceId, + }); + } catch (error) { + return problemResponse(error, traceId); + } +} diff --git a/apps/web/app/api/v1/tasks/[id]/cancel/route.ts b/apps/web/app/api/v1/tasks/[id]/cancel/route.ts index 3eb1ef6..9d63979 100644 --- a/apps/web/app/api/v1/tasks/[id]/cancel/route.ts +++ b/apps/web/app/api/v1/tasks/[id]/cancel/route.ts @@ -7,6 +7,7 @@ import { problemResponse, requestTraceId, } from "@/lib/api-context"; +import { agentGatewayHeaders } from "@/lib/agent-gateway"; import { settleAgentRun } from "@/lib/task-domain"; export async function POST( @@ -49,6 +50,7 @@ export async function POST( const gateway = await fetch( `${process.env.AGENT_GATEWAY_URL ?? "http://agent-gateway:3002"}/v1/runs/${encodeURIComponent(task.agentRunId)}/cancel`, { + headers: agentGatewayHeaders(subject.organisationId), method: "POST", signal: AbortSignal.timeout(5_000), }, diff --git a/apps/web/app/api/v1/tasks/[id]/events/route.ts b/apps/web/app/api/v1/tasks/[id]/events/route.ts index ac4ca64..77e0288 100644 --- a/apps/web/app/api/v1/tasks/[id]/events/route.ts +++ b/apps/web/app/api/v1/tasks/[id]/events/route.ts @@ -7,6 +7,7 @@ import { problemResponse, requestTraceId, } from "@/lib/api-context"; +import { agentGatewayHeaders } from "@/lib/agent-gateway"; import { settleAgentRun, type AgentRunResult } from "@/lib/task-domain"; export const dynamic = "force-dynamic"; @@ -47,7 +48,10 @@ export async function GET( while (!request.signal.aborted) { const gateway = await fetch( `${process.env.AGENT_GATEWAY_URL ?? "http://agent-gateway:3002"}/v1/runs/${encodeURIComponent(task.agentRunId!)}`, - { signal: AbortSignal.timeout(5_000) }, + { + headers: agentGatewayHeaders(subject.organisationId), + signal: AbortSignal.timeout(5_000), + }, ); const result = (await gateway.json()) as AgentRunResult & { status: "running" | "completed" | "failed" | "cancelled"; diff --git a/apps/web/app/api/v1/tasks/route.ts b/apps/web/app/api/v1/tasks/route.ts index c286b65..09b2123 100644 --- a/apps/web/app/api/v1/tasks/route.ts +++ b/apps/web/app/api/v1/tasks/route.ts @@ -1,4 +1,4 @@ -import { and, asc, eq, inArray } from "drizzle-orm"; +import { and, asc, eq, inArray, isNull } from "drizzle-orm"; import { requireCapability } from "@muster/authz"; import { redactForObservation } from "@muster/config"; import { TaskPrioritySchema, TaskStatusSchema } from "@muster/contracts"; @@ -28,7 +28,12 @@ async function taskView(organisationId: string, includeEvidence: boolean) { const rows = await db .select() .from(schema.tasks) - .where(eq(schema.tasks.organisationId, organisationId)) + .where( + and( + eq(schema.tasks.organisationId, organisationId), + isNull(schema.tasks.archivedAt), + ), + ) .orderBy(asc(schema.tasks.status), asc(schema.tasks.createdAt)); const actorIds = rows .map((task) => task.assignedActorId) diff --git a/apps/web/app/settings/slack/page.tsx b/apps/web/app/settings/slack/page.tsx new file mode 100644 index 0000000..37dcd6b --- /dev/null +++ b/apps/web/app/settings/slack/page.tsx @@ -0,0 +1,5 @@ +import { SlackSettingsView } from "@/components/slack-settings-view"; + +export default function SlackSettingsPage() { + return ; +} diff --git a/apps/web/components/agent-run-view.tsx b/apps/web/components/agent-run-view.tsx new file mode 100644 index 0000000..17bff16 --- /dev/null +++ b/apps/web/components/agent-run-view.tsx @@ -0,0 +1,112 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { AppShell } from "@/components/app-shell"; +import { PageHeader } from "@/components/page-header"; +import { Badge } from "@/components/ui/badge"; + +type HarnessRun = { + protocolVersion: "muster.agent-harness/v1"; + runId: string; + status: string; + agentKey: string; + correlationId: string; + duplicate: boolean; + result: unknown; +}; + +export function AgentRunView({ runId }: { runId: string }) { + const [run, setRun] = useState(null); + const [error, setError] = useState(""); + + useEffect(() => { + const controller = new AbortController(); + void fetch(`/api/v1/agent-harness/runs/${encodeURIComponent(runId)}`, { + cache: "no-store", + signal: controller.signal, + }) + .then(async (response) => { + const body = (await response.json()) as { + data?: HarnessRun; + detail?: string; + }; + if (!response.ok || !body.data) + throw new Error(body.detail ?? "Agent run is unavailable."); + setRun(body.data); + }) + .catch((cause: unknown) => { + if (!controller.signal.aborted) + setError( + cause instanceof Error + ? cause.message + : "Agent run is unavailable.", + ); + }); + return () => controller.abort(); + }, [runId]); + + return ( + + +
+
+ {run && ( + <> +
+
+

+ {run.agentKey} +

+ {run.status} +
+
+
+
+ Run +
+
+ {run.runId} +
+
+
+
+ Correlation +
+
+ {run.correlationId} +
+
+
+
+
+

Typed result

+ 
+                  {run.result === null
+                    ? "No typed result is available yet."
+                    : JSON.stringify(run.result, null, 2)}
+
+
+ + )} + {!run && !error && ( +

+ Loading agent run… +

+ )} + {error && ( +

+ {error} +

+ )} +
+
+ + ); +} diff --git a/apps/web/components/page-header.tsx b/apps/web/components/page-header.tsx index 37624e3..765972c 100644 --- a/apps/web/components/page-header.tsx +++ b/apps/web/components/page-header.tsx @@ -12,7 +12,7 @@ export function PageHeader({ actions?: ReactNode; }) { return ( -
+
{eyebrow && (

@@ -27,7 +27,9 @@ export function PageHeader({ )}

{actions && ( -
{actions}
+
+ {actions} +
)}
); diff --git a/apps/web/components/slack-settings-view.tsx b/apps/web/components/slack-settings-view.tsx new file mode 100644 index 0000000..c9fcdd4 --- /dev/null +++ b/apps/web/components/slack-settings-view.tsx @@ -0,0 +1,848 @@ +"use client"; + +import Link from "next/link"; +import { + type KeyboardEvent, + useCallback, + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import { AppShell } from "@/components/app-shell"; +import { PageHeader } from "@/components/page-header"; +import { Button } from "@/components/ui/button"; + +type Installation = { + id: string; + teamId: string; + teamName: string | null; + scopes: unknown; + status: string; + installedAt: string; + lastHealthAt: string | null; + lastDeliveryAt: string | null; + lastError: string | null; +}; + +type Actor = { id: string; displayName: string }; +type Agent = { id: string; name: string }; +type Identity = { + id: string; + installationId: string; + slackUserId: string; + actorId: string; + actorName: string; + status: string; + createdAt: string; +}; +type Exposure = { + id: string; + installationId: string; + agentId: string; + agentName: string; + enabled: boolean; + isDefault: boolean; + allowedChannelIds: unknown; + allowDirectMessages: boolean; + allowThreadContext: boolean; + updatedAt: string; +}; +type Delivery = { + id: string; + installationId: string; + runId: string; + status: string; + attemptCount: number; + lastError: string | null; + updatedAt: string; +}; +type SlackSettings = { + installations: Installation[]; + actors: Actor[]; + agents: Agent[]; + identities: Identity[]; + exposures: Exposure[]; + deliveries: Delivery[]; +}; + +const date = (value: string | null) => + value ? new Date(value).toLocaleString() : "Not recorded"; + +const channels = (value: unknown) => + Array.isArray(value) + ? value.filter((item): item is string => typeof item === "string") + : []; + +export function SlackSettingsView() { + const [settings, setSettings] = useState(null); + const [error, setError] = useState(null); + const [notice, setNotice] = useState(null); + const [loading, setLoading] = useState(true); + const [busy, setBusy] = useState(null); + const [exposureEnabled, setExposureEnabled] = useState(true); + const [exposureDefault, setExposureDefault] = useState(false); + const [revokeCandidate, setRevokeCandidate] = useState( + null, + ); + const hasLoaded = useRef(false); + const reconnectButton = useRef(null); + const revokeDialog = useRef(null); + const revokeCancelButton = useRef(null); + const revokeReturnFocus = useRef(null); + + const load = useCallback(async () => { + setLoading(!hasLoaded.current); + try { + const response = await fetch("/api/v1/slack/settings", { + cache: "no-store", + }); + const payload = (await response.json()) as { + data?: SlackSettings; + detail?: string; + }; + if (!response.ok || !payload.data) + throw new Error( + payload.detail ?? "Could not load Slack administration settings.", + ); + setSettings(payload.data); + setError(null); + hasLoaded.current = true; + } catch (cause) { + setError( + cause instanceof Error + ? cause.message + : "Could not load Slack administration settings.", + ); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => void load(), [load]); + + useEffect(() => { + if (!revokeCandidate) return; + const frame = requestAnimationFrame(() => + revokeCancelButton.current?.focus(), + ); + return () => cancelAnimationFrame(frame); + }, [revokeCandidate]); + + const activeInstallations = useMemo( + () => + settings?.installations.filter( + (installation) => installation.status === "active", + ) ?? [], + [settings], + ); + + const reconnect = async () => { + setBusy("reconnect"); + setNotice(null); + setError(null); + try { + const response = await fetch("/api/v1/slack/install", { + cache: "no-store", + }); + const payload = (await response.json()) as { + data?: { authorizationUrl?: string }; + detail?: string; + }; + if (!response.ok || !payload.data?.authorizationUrl) + throw new Error(payload.detail ?? "Could not start Slack OAuth."); + window.location.assign(payload.data.authorizationUrl); + } catch (cause) { + setError( + cause instanceof Error ? cause.message : "Could not start Slack OAuth.", + ); + } finally { + setBusy(null); + } + }; + + const refreshHealth = async () => { + setBusy("health"); + setNotice(null); + setError(null); + try { + const response = await fetch("/api/v1/slack/health", { + cache: "no-store", + }); + const payload = (await response.json()) as { detail?: string }; + if (!response.ok) + throw new Error(payload.detail ?? "Slack health refresh failed."); + await load(); + setNotice("Slack diagnostics refreshed."); + } catch (cause) { + setError( + cause instanceof Error ? cause.message : "Slack health refresh failed.", + ); + } finally { + setBusy(null); + } + }; + + const request = async ( + url: string, + method: "POST" | "PUT", + body: unknown, + ) => { + const response = await fetch(url, { + method, + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); + const payload = (await response.json()) as { detail?: string }; + if (!response.ok) + throw new Error(payload.detail ?? "Slack administration update failed."); + }; + + const saveIdentity = async (form: HTMLFormElement) => { + const values = new FormData(form); + setBusy("identity"); + setNotice(null); + setError(null); + try { + await request("/api/v1/slack/identities", "POST", { + installationId: values.get("installationId"), + slackUserId: values.get("slackUserId"), + actorId: values.get("actorId"), + }); + form.reset(); + await load(); + setNotice("Slack user mapping saved."); + } catch (cause) { + setError( + cause instanceof Error + ? cause.message + : "Could not save Slack user mapping.", + ); + } finally { + setBusy(null); + } + }; + + const saveExposure = async (form: HTMLFormElement) => { + const values = new FormData(form); + const allowedChannelIds = String(values.get("allowedChannelIds") ?? "") + .split(/[\n,]/) + .map((value) => value.trim()) + .filter(Boolean); + const enabled = values.get("enabled") === "on"; + setBusy("exposure"); + setNotice(null); + setError(null); + try { + await request("/api/v1/slack/exposures", "PUT", { + installationId: values.get("installationId"), + agentId: values.get("agentId"), + enabled, + isDefault: enabled && values.get("isDefault") === "on", + allowedChannelIds, + allowDirectMessages: values.get("allowDirectMessages") === "on", + allowThreadContext: values.get("allowThreadContext") === "on", + }); + await load(); + setNotice("Agent exposure policy saved."); + } catch (cause) { + setError( + cause instanceof Error + ? cause.message + : "Could not save agent exposure policy.", + ); + } finally { + setBusy(null); + } + }; + + const revoke = async () => { + if (!revokeCandidate) return; + setBusy("revoke"); + setNotice(null); + setError(null); + let completed = false; + try { + const response = await fetch( + `/api/v1/slack/install?installationId=${encodeURIComponent(revokeCandidate.id)}`, + { method: "DELETE" }, + ); + const payload = (await response.json()) as { detail?: string }; + if (!response.ok) + throw new Error( + payload.detail ?? "Could not revoke Slack installation.", + ); + setRevokeCandidate(null); + await load(); + setNotice("Slack installation revoked. Existing tokens were replaced."); + completed = true; + } catch (cause) { + setError( + cause instanceof Error + ? cause.message + : "Could not revoke Slack installation.", + ); + } finally { + setBusy(null); + if (completed) + requestAnimationFrame(() => reconnectButton.current?.focus()); + } + }; + + const closeRevoke = () => { + const returnTarget = revokeReturnFocus.current; + setRevokeCandidate(null); + requestAnimationFrame(() => { + if (returnTarget?.isConnected) returnTarget.focus(); + else reconnectButton.current?.focus(); + }); + }; + + const handleRevokeKeyDown = (event: KeyboardEvent) => { + if (event.key === "Escape" && busy === null) { + event.preventDefault(); + closeRevoke(); + return; + } + if (event.key !== "Tab") return; + const controls = Array.from( + revokeDialog.current?.querySelectorAll( + 'button:not([disabled]), [href], input:not([disabled]), select:not([disabled]), textarea:not([disabled]), [tabindex]:not([tabindex="-1"])', + ) ?? [], + ); + const first = controls[0]; + const last = controls.at(-1); + if (!first || !last) return; + if (event.shiftKey && document.activeElement === first) { + event.preventDefault(); + last.focus(); + } else if (!event.shiftKey && document.activeElement === last) { + event.preventDefault(); + first.focus(); + } + }; + + return ( + + + + + + } + /> + +
+ {error ? ( +
+

{error}

+ {!settings ? ( + + ) : null} +
+ ) : null} + {notice ? ( +

+ {notice} +

+ ) : null} + {loading ? ( +

+ Loading Slack administration… +

+ ) : null} + + {!loading && settings ? ( + <> +
+
+
+

+ Workspace connections +

+

+ Tokens stay encrypted; this view only shows redacted + operator diagnostics. +

+
+
+ {settings.installations.length === 0 ? ( +

+ No Slack workspace is connected. Connect Slack to begin a + governed installation. +

+ ) : null} +
+ {settings.installations.map((installation) => ( +
+
+

+ {installation.teamName ?? installation.teamId} +

+

+ {installation.status} · installed{" "} + {date(installation.installedAt)} +

+

+ Health: {date(installation.lastHealthAt)} · latest + delivery: {date(installation.lastDeliveryAt)} +

+

+ Scopes:{" "} + {channels(installation.scopes).join(", ") || + "Not recorded"} +

+ {installation.lastError ? ( +

+ {installation.lastError} +

+ ) : null} +
+ {installation.status === "active" ? ( + + ) : null} +
+ ))} +
+
+ +
+
{ + event.preventDefault(); + void saveIdentity(event.currentTarget); + }} + > +

+ Map a Slack user +

+

+ Map one Slack user to one active Muster human. Approvals still + require the authoritative Muster capability checks. +

+ {activeInstallations.length === 0 ? ( +

+ Connect an active Slack workspace before mapping users. +

+ ) : settings.actors.length === 0 ? ( +

+ No active Muster humans are available for identity mapping. +

+ ) : null} + + + + +
+ {settings.identities.length === 0 ? ( +

+ No Slack identities are mapped yet. +

+ ) : ( + settings.identities.map((identity) => ( +

+ + {identity.slackUserId} + {" "} + → {identity.actorName} · {identity.status} +

+ )) + )} +
+
+ +
{ + event.preventDefault(); + void saveExposure(event.currentTarget); + }} + > +

+ Agent exposure policy +

+

+ Channel mentions are allowed only for listed channel IDs. + Direct-message and thread-context access are explicit per + agent. +

+

+ Emergency execution control remains on each agent's{" "} + {settings.agents[0] ? ( + + kill switch + + ) : ( + "kill switch" + )} + . +

+ {activeInstallations.length === 0 ? ( +

+ Connect an active Slack workspace before exposing agents. +

+ ) : settings.agents.length === 0 ? ( +

+ No active agents are available. Restore or configure an + agent before creating a Slack policy. +

+ ) : null} + + +