commit bc55924198865502203a992b6ea914a38fd9d173 Author: Codex Lead Engineer Date: Thu Jul 30 02:13:26 2026 +0200 feat: implement project bus MVP diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..574c9e0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +__pycache__/ +*.py[cod] +*.db +*.db-shm +*.db-wal +.coverage +.pytest_cache/ +.venv/ +dist/ +build/ +*.egg-info/ +.env + diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..2125b2b --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,17 @@ +# Changelog + +All notable changes to this project are documented here. + +## 0.1.0 — 2026-07-30 + +Initial review candidate: + +- generic project, actor-role, work-package, result, review, decision, + escalation, gate, baseline, and event models; +- append-only SQLite event log with cursor synchronization; +- concurrency-safe and idempotent mutation handling; +- stateless MCP Streamable HTTP tool interface; +- transport authentication abstraction and BASDM separation controls; +- tests, container example, contract, threat model, production profile, and + WP-PA-001 review evidence. + diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..ee0c332 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Epimonos contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..fdd97d2 --- /dev/null +++ b/README.md @@ -0,0 +1,167 @@ +# Epimonos Project Bus + +Epimonos Project Bus is a small, generic coordination service for auditable +human/agent projects. It exposes a stateless MCP Streamable HTTP endpoint and +keeps an ordered, append-only event trail alongside relational projections. + +The core has no Sandbox, Epimonos product, billing, identity-provider, or +backoffice domain logic. A project supplies its own actors, role assignments, +work-package contents, decisions, evidence URIs, and repository references. + +## MVP capabilities + +- isolated projects and actors with one formal role per project; +- issued, atomically claimed, and evidenced work packages; +- independent review requests and verdicts; +- product, architecture, operational, and gate decisions; +- escalations and formal resolution; +- commit and artifact references; +- PM/PO gate closure and baseline-release events; +- ordered cursor-based event synchronization; +- idempotent mutations and concurrency-safe local writes; +- transport-derived identity and role-based authorization; +- database-level protection against event update and deletion. + +## Governance invariants + +The service enforces these minimum separations: + +- a PA cannot record product, architecture, operational, or gate decisions; +- only a PO can record a product decision; +- only a PM or PO can close a gate or release a baseline; +- the actor who claimed/submitted a result cannot review it; +- a gate cannot accept a result without an independent `APPROVE` review; +- an actor who submitted a result cannot accept it; +- a baseline can reference only accepted work packages. + +These are technical guardrails, not a replacement for the project role +contracts or the PM's tolerance and escalation policy. + +## Quick start + +Python 3.12 is the only runtime dependency. + +```bash +export PYTHONPATH=src +export PROJECT_BUS_BOOTSTRAP_TOKEN='replace-this-with-long-secret' +python -m project_bus --database ./project-bus.db --host 127.0.0.1 --port 8080 +``` + +Health endpoint: + +```bash +curl http://127.0.0.1:8080/healthz +``` + +Initialize MCP: + +```bash +curl http://127.0.0.1:8080/mcp \ + -H 'Content-Type: application/json' \ + --data '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"curl","version":"1"}}}' +``` + +Create a project: + +```bash +curl http://127.0.0.1:8080/mcp \ + -H 'Content-Type: application/json' \ + -H 'Authorization: Bearer replace-this-with-long-secret' \ + --data '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"create_project","arguments":{"project_id":"example","name":"Example","idempotency_key":"bootstrap-example-v1"}}}' +``` + +Use `register_actor` with the bootstrap identity to appoint a PO. Configure +the actor's bearer credential through `PROJECT_BUS_TOKENS_JSON`, then let the +PO/PM register the remaining members. Actor identity is always taken from the +authenticated bearer credential; callers cannot supply an acting identity in +tool arguments. + +Example credential configuration: + +```bash +export PROJECT_BUS_TOKENS_JSON='{ + "long-random-po-token": { + "actor_id": "po-1", + "auth_subject": "po-1", + "display_name": "Product Owner" + } +}' +``` + +The identity must match an actor registered in the target project. Environment +changes require a process restart in this MVP. + +## MCP client configuration + +For Claude Code: + +```bash +claude mcp add \ + --scope project \ + --transport http \ + epimonos-project-bus \ + https://project-bus.example/mcp +``` + +For a local-only development server, use the local HTTP URL. Credentials +belong in the client/user secret store, never in `.mcp.json` or Git. + +## Configuration + +| Variable | Default | Meaning | +|---|---:|---| +| `PROJECT_BUS_DATABASE` | `project-bus.db` | SQLite database path | +| `PROJECT_BUS_MIGRATIONS` | source-tree discovery | Migration directory | +| `PROJECT_BUS_HOST` | `127.0.0.1` | Bind address | +| `PROJECT_BUS_PORT` | `8080` | HTTP port | +| `PROJECT_BUS_BOOTSTRAP_TOKEN` | required, minimum 24 characters | SYSTEM bearer token | +| `PROJECT_BUS_TOKENS_JSON` | empty | Static bearer-token identity map | +| `PROJECT_BUS_ALLOWED_ORIGINS` | empty | Comma-separated exact browser origins | +| `PROJECT_BUS_LOG_LEVEL` | `INFO` | Python log level | + +Never expose the bootstrap token. On an internet-facing deployment, terminate +TLS at a trusted reverse proxy and set a strict origin allowlist. + +## Test + +```bash +PYTHONPATH=src python -m unittest discover -v +``` + +The suite uses only temporary databases and includes concurrent writer tests. + +## Local persistence and production profile + +SQLite runs in WAL mode and every mutation uses `BEGIN IMMEDIATE`. This gives +a dependable single-process/single-node MVP and prevents claim and invariant +check races. The append-only event table is protected with `UPDATE` and +`DELETE` rejection triggers. + +Production differences are intentionally explicit: + +| Concern | MVP | Production target | +|---|---|---| +| Database | SQLite/WAL, one node | PostgreSQL 16+, transactions and row locks | +| Identity | static bearer map | OIDC/JWT or mTLS workload identity | +| Secrets | environment | secret manager, rotation, revocation | +| Transport | plain HTTP capable | TLS-only reverse proxy/service mesh | +| Scaling | one server process | multiple stateless MCP instances | +| Audit durability | local DB/backup | PITR, replicas, external immutable export | +| Availability | process supervisor | orchestrator probes, SLOs, alerting | +| Abuse control | body cap/origin policy | edge rate limits, quotas, WAF where useful | + +The persistence boundary is deliberately small (`Database` plus SQL in the +service), but this release does **not** contain a PostgreSQL adapter. SQLite +must not be mounted on NFS or used for active-active replicas. + +## Repository guide + +- `src/project_bus/` — domain service, auth abstraction, MCP and HTTP adapter; +- `migrations/` — ordered relational schema; +- `tests/` — service, concurrency, governance, and HTTP contract tests; +- `docs/` — architecture, tool contract, threat model, production notes; +- `deploy/` — container and Compose example; +- `evidence/` — WP-PA-001 verification and independent-review handoff. + +See [docs/TOOL_CONTRACT.md](docs/TOOL_CONTRACT.md) before integrating a new +actor and [docs/THREAT_MODEL.md](docs/THREAT_MODEL.md) before deployment. diff --git a/deploy/.env.example b/deploy/.env.example new file mode 100644 index 0000000..995500d --- /dev/null +++ b/deploy/.env.example @@ -0,0 +1,4 @@ +PROJECT_BUS_BOOTSTRAP_TOKEN=replace-with-a-long-random-secret +PROJECT_BUS_TOKENS_JSON={} +PROJECT_BUS_ALLOWED_ORIGINS=https://chatgpt.com + diff --git a/deploy/Dockerfile b/deploy/Dockerfile new file mode 100644 index 0000000..2304f16 --- /dev/null +++ b/deploy/Dockerfile @@ -0,0 +1,23 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PROJECT_BUS_DATABASE=/data/project-bus.db \ + PROJECT_BUS_MIGRATIONS=/app/migrations \ + PROJECT_BUS_HOST=0.0.0.0 \ + PROJECT_BUS_PORT=8080 + +WORKDIR /app +COPY pyproject.toml README.md LICENSE ./ +COPY src ./src +COPY migrations ./migrations +RUN pip install --no-cache-dir --no-deps . \ + && mkdir -p /data \ + && chown -R 65532:65532 /data + +USER 65532:65532 +EXPOSE 8080 +VOLUME ["/data"] +HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \ + CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/healthz', timeout=2)"] +ENTRYPOINT ["project-bus"] diff --git a/deploy/compose.yaml b/deploy/compose.yaml new file mode 100644 index 0000000..846fd2e --- /dev/null +++ b/deploy/compose.yaml @@ -0,0 +1,25 @@ +services: + project-bus: + build: + context: .. + dockerfile: deploy/Dockerfile + restart: unless-stopped + environment: + PROJECT_BUS_BOOTSTRAP_TOKEN: "${PROJECT_BUS_BOOTSTRAP_TOKEN:?set a strong bootstrap token}" + PROJECT_BUS_TOKENS_JSON: "${PROJECT_BUS_TOKENS_JSON:-}" + PROJECT_BUS_ALLOWED_ORIGINS: "${PROJECT_BUS_ALLOWED_ORIGINS:-}" + ports: + - "127.0.0.1:8080:8080" + volumes: + - project-bus-data:/data + read_only: true + tmpfs: + - /tmp:size=16m,noexec,nosuid,nodev + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + +volumes: + project-bus-data: + diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md new file mode 100644 index 0000000..fe83dc8 --- /dev/null +++ b/docs/ARCHITECTURE.md @@ -0,0 +1,61 @@ +# Architecture + +## Boundary + +The Project Bus coordinates formal project events. It does not execute source +code, clone repositories, inspect artifacts, send chat messages, implement a +project methodology, or decide what a project should build. Those activities +remain with authenticated actors and external systems. + +## Components + +1. `ProjectBusRequestHandler` implements a small stateless MCP Streamable HTTP + transport over JSON-RPC 2.0. +2. `AuthProvider` turns transport credentials into an immutable principal. +3. `MCPApplication` publishes tools and maps calls to domain operations. +4. `ProjectBusService` enforces role, lifecycle, independence, idempotency, + and gate invariants. +5. `Database` owns migration and transaction boundaries. +6. SQLite relational projections support current-state queries; `event_log` + is the immutable ordered synchronization source. + +## Write path + +A mutating call is authenticated before tool dispatch. The service opens +`BEGIN IMMEDIATE`, resolves the principal to an active project membership, +checks the idempotency tuple `(project, actor, key)`, validates current state, +updates the relevant projection, appends exactly one event, stores the +canonical response, and commits. + +The event and projection update therefore succeed or fail together. +Idempotency replays return the original response without adding an event. + +## Event ordering and cursors + +SQLite allocates a monotonic integer `cursor` for each event. `sync_since` +filters by project, returns ascending cursors, and returns the last delivered +cursor. Consumers persist that value only after processing the response. The +cursor is global to the database, so gaps within a project's stream are +normal. Consumers must not infer missing project events from gaps. + +This is pull synchronization. The MCP server does not wake a ChatGPT thread or +push into a dormant CLI session. + +## State model + +Work packages move through: + +`OPEN → CLAIMED → RESULT_SUBMITTED → IN_REVIEW → REVIEWED → ACCEPTED|REJECTED` + +`HOLD` leaves a reviewed package in `REVIEWED`. A new review can be requested +after a completed review. This MVP treats corrections after +`CHANGES_REQUESTED` as a new work package or a new explicitly governed +iteration; it does not silently overwrite submitted result evidence. + +## Trust boundaries + +- Authentication proves an external subject; project membership grants a role. +- The database is trusted to enforce integrity and append-only triggers. +- Artifact URIs and commit hashes are references, not trusted content. +- Reverse proxy identity headers are not consumed by the built-in adapter. + diff --git a/docs/PRODUCTION_PROFILE.md b/docs/PRODUCTION_PROFILE.md new file mode 100644 index 0000000..a519ab4 --- /dev/null +++ b/docs/PRODUCTION_PROFILE.md @@ -0,0 +1,36 @@ +# Production Profile + +The MVP is suitable for local integration and controlled proof-of-concept use. +Internet-facing or business-critical deployment requires a separate production +hardening work package. + +## Required changes + +1. Implement the persistence interface for PostgreSQL 16+. + Use transactions, `SELECT … FOR UPDATE` or conditional updates, unique + idempotency constraints, and database roles that cannot update/delete the + event table. +2. Replace `StaticTokenAuthProvider` with validated OIDC/JWT or mTLS workload + identity. Bind issuer, audience, subject, expiry, and revocation policy. +3. Put the service behind TLS and authenticated ingress. Do not trust arbitrary + forwarded identity headers. +4. Run multiple stateless instances only after PostgreSQL conformance and + concurrency tests pass. +5. Add structured audit export, metrics, tracing correlation, alerts, backup, + point-in-time recovery, and restore exercises. +6. Add membership lifecycle, token revocation, retention/export policy, and + project deletion/archival decisions. Event deletion is intentionally absent. +7. Pin and scan the base image; generate an SBOM and sign release artifacts. +8. Run load, fault, client interoperability, and penetration tests. + +## MCP activation + +The deployed endpoint must support `POST /mcp` with `application/json`. The +current service returns normal JSON instead of SSE and uses no session IDs. +Validate that each intended client accepts this stateless profile before +production gate acceptance. + +Chat threads are not push subscribers. A client, automation, or PA invocation +must call `sync_since(last_cursor)`. The cursor itself belongs in durable client +state and is advanced only after successful processing. + diff --git a/docs/THREAT_MODEL.md b/docs/THREAT_MODEL.md new file mode 100644 index 0000000..98d7f9f --- /dev/null +++ b/docs/THREAT_MODEL.md @@ -0,0 +1,83 @@ +# Threat Model + +## Assets + +- project instructions, decisions, results, reviews, and escalations; +- actor-role assignments and identity bindings; +- event ordering and audit integrity; +- bearer credentials; +- commit, artifact, and baseline references; +- service/database availability. + +## Adversaries + +- an unauthenticated network client; +- an authenticated actor exceeding their formal mandate; +- an implementer attempting self-review or self-acceptance; +- a compromised bearer token; +- a malicious artifact URI or oversized request; +- an operator or database account attempting audit-history mutation; +- concurrent clients exploiting check/write races. + +## Implemented controls + +| Threat | MVP control | +|---|---| +| Identity spoofing in tool args | no actor argument; identity comes from transport | +| Token timing comparison | SHA-256 digest and constant-time comparison | +| Unauthorized project access | active membership lookup per call | +| PA decision overreach | explicit decision/gate/baseline denial | +| Product decision overreach | PO-only product decisions | +| Self-review | reviewer role plus actor independence checks | +| Self-acceptance | PM/PO role and result-submitter identity check | +| Premature gate acceptance | independent APPROVE review required | +| Double claim / TOCTOU | `BEGIN IMMEDIATE` and conditional state update | +| Duplicate delivery | scoped request hash and idempotency record | +| Audit update/delete | SQLite triggers reject both operations | +| Cross-project reads | project filter plus membership resolution | +| Browser cross-origin request | exact configurable Origin allowlist | +| Oversized request | one MiB body limit | +| MIME confusion/caching | strict JSON input, nosniff, no-store | +| SQL injection | parameterized SQL; one controlled placeholder expansion | + +Artifact URIs and review evidence are recorded as untrusted data and are never +fetched or executed by the bus. + +## Residual risks and production requirements + +1. Static bearer credentials have no built-in expiry, rotation endpoint, + audience, issuer, or proof-of-possession. Replace the adapter with validated + OIDC/JWT or mTLS identities. +2. Plain HTTP is available. Use TLS and prohibit direct public access to the + application port. +3. SQLite append-only triggers do not protect against an operator replacing + the entire database file or disabling triggers. Use restricted DB roles, + PostgreSQL permissions, PITR, backups, and an external immutable audit + export for production assurance. +4. The event payload is not cryptographically chained or signed. Add hash + chaining/signatures only if the agreed threat model requires tamper + evidence against privileged database operators. +5. No application rate limiter exists. Apply per-identity limits and request + quotas at the trusted edge. +6. The server emits access metadata but no security audit sink or metrics. + Integrate structured logs, alerts, and privacy-aware retention. +7. Membership deactivation and credential revocation are not exposed as MVP + tools. Operators must treat compromised tokens as an immediate + configuration/restart incident. +8. Role assignment is one role per actor/project. Shared human/agent accounts + undermine independence even when IDs differ and are prohibited operationally. +9. Evidence and artifact digests are caller assertions; the bus does not + verify repository ownership or retrieve content. +10. Dependency-free MCP handling reduces supply-chain surface but has not yet + been interoperability-tested against every intended client version. + +## Security verification before activation + +- run all tests and an independent code review; +- test the exact Claude Code and ChatGPT MCP clients; +- replace all example tokens and restrict origin/host exposure; +- confirm backup/restore and audit export; +- review log contents for sensitive project data; +- add abuse limits and operational monitoring; +- decide whether privileged-operator tamper evidence is required. + diff --git a/docs/TOOL_CONTRACT.md b/docs/TOOL_CONTRACT.md new file mode 100644 index 0000000..6a4edb6 --- /dev/null +++ b/docs/TOOL_CONTRACT.md @@ -0,0 +1,79 @@ +# MCP Tool Contract + +Protocol version: `2025-03-26` +Endpoint: `POST /mcp` +Transport profile: stateless Streamable HTTP with JSON responses + +`initialize`, `ping`, and `tools/list` do not require credentials. Every +`tools/call` requires `Authorization: Bearer …`. The server does not implement +server-to-client SSE notifications in this MVP; `GET /mcp` returns `405`. + +## Common rules + +- Every project-scoped call resolves the authenticated principal to one active + membership. +- Every mutation requires a stable `idempotency_key`. +- Reusing a key with the same operation and canonical arguments replays the + original response. Reusing it with different input returns `conflict`. +- IDs use letters, numbers, `.`, `_`, `:`, `/`, or `-`, up to 128 characters. +- Tool domain failures are MCP tool results with `isError: true`. +- JSON-RPC syntax/method failures use JSON-RPC error objects. +- Missing or invalid credentials return HTTP `401`. + +## Tools + +| Tool | Allowed role / invariant | Primary event | +|---|---|---| +| `create_project` | SYSTEM only | `project.created` | +| `register_actor` | SYSTEM, PO, PM; PM cannot appoint PO | `actor.registered` | +| `issue_work_package` | PO, PM | `work_package.issued` | +| `claim_work_package` | Architect, Lead Engineer, Implementer; atomic OPEN claim | `work_package.claimed` | +| `submit_result` | claiming actor only | `work_package.result_submitted` | +| `request_review` | PO, PM, PA, implementation/design roles; reviewer independent | `review.requested` | +| `submit_review` | assigned independent review actor | `review.submitted` | +| `record_decision` | product: PO; other kinds: PO/PM; never PA | `decision.recorded` | +| `raise_escalation` | every active member | `escalation.raised` | +| `resolve_escalation` | PO, PM | `escalation.resolved` | +| `link_commit` | every active member; valid hexadecimal object ID | `commit.linked` | +| `link_artifact` | every active member | `artifact.linked` | +| `close_gate` | PO, PM; ACCEPT needs independent approval | `gate.closed` | +| `baseline_release` | PO, PM; included work packages accepted | `baseline.released` | +| `sync_since` | every active member; max 500 events | none | +| `list_pending_actions` | every active member, role-filtered reviews/escalations | none | +| `get_work_package` | every active member | none | +| `get_project` | every active member; includes actor-role roster | none | + +The authoritative argument schema is returned by `tools/list`. Clients should +discover it instead of hard-coding optional defaults. + +## Event envelope + +`sync_since` returns: + +```json +{ + "cursor": 42, + "event_id": "evt_...", + "project_id": "example", + "event_type": "review.submitted", + "actor_id": "reviewer-1", + "actor_role": "REVIEWER", + "aggregate_type": "review", + "aggregate_id": "rev_...", + "correlation_id": "WP-001", + "causation_event_id": null, + "created_at": "2026-07-30T00:00:00.000Z", + "payload": {} +} +``` + +Event payloads are version-zero MVP structures. Consumers should ignore +unknown fields and key behavior off `event_type`, not prose. + +## MCP interoperability limits + +The transport implements the required request/response subset used by MCP +tools over Streamable HTTP. It is stateless, does not issue `Mcp-Session-Id`, +does not accept JSON-RPC batches, and provides no resumable SSE channel. +Interoperability must be checked against the concrete ChatGPT and Claude Code +clients during activation. diff --git a/evidence/CLAUDE_REVIEW_BRIEF.md b/evidence/CLAUDE_REVIEW_BRIEF.md new file mode 100644 index 0000000..f44dc91 --- /dev/null +++ b/evidence/CLAUDE_REVIEW_BRIEF.md @@ -0,0 +1,45 @@ +# Independent Claude Review Brief + +## Role + +Act as independent Lead Reviewer for `WP-PA-001 — Epimonos Project Bus MVP`. +Do not modify the implementation during the first review pass. The implementer +is Codex; reviewer identity and evidence must remain independent. + +## Review objectives + +1. Verify every requested capability against code, schema, tests, and docs. +2. Challenge BASDM separations: + - PA cannot make product/architecture/gate decisions; + - implementer cannot review or accept own work; + - only authorized PM/PO actors close gates and release baselines. +3. Inspect transaction boundaries, idempotency replay, concurrent claims, + cursor semantics, and projection/event atomicity. +4. Assess authentication, cross-project isolation, origin handling, input + validation, SQL usage, audit immutability, and operational failure modes. +5. Check MCP `2025-03-26` request/response interoperability assumptions. +6. Confirm the core contains no Sandbox/Epimonos product-specific logic. +7. Confirm documentation accurately states all production gaps. + +## Required output + +Return: + +- verdict: `APPROVE`, `CHANGES_REQUESTED`, or `REJECT`; +- numbered findings with severity `Critical/High/Medium/Low`; +- exact file and line reference; +- exploit/failure scenario; +- required correction and suggested verification; +- scope-completeness matrix; +- residual-risk statement. + +Run at minimum: + +```bash +PYTHONPATH=src python -m unittest discover -v +PYTHONPATH=src python -m compileall -q src tests +``` + +Also add adversarial tests where a claimed invariant is not adequately proven. +Do not accept solely because the existing suite is green. + diff --git a/evidence/VERIFICATION.txt b/evidence/VERIFICATION.txt new file mode 100644 index 0000000..c57147b --- /dev/null +++ b/evidence/VERIFICATION.txt @@ -0,0 +1,40 @@ +WP-PA-001 VERIFICATION +Date: 2026-07-30 +Candidate version: 0.1.0 + +RESULTS +- Python source compilation: PASS +- Source-tree test suite: PASS (16 tests) +- Built/installed-package test suite: PASS (16 tests) +- Installed-package migration smoke test: PASS +- Local Git repository root initialized: PASS +- External repository creation/push: NOT PERFORMED (by scope) +- Container build: NOT RUN (Docker unavailable in execution environment) +- Independent Claude review: PENDING +- ChatGPT/Claude Code client interoperability: PENDING ACTIVATION + +COMMANDS +PYTHONPATH=src python -m compileall -q src tests +PYTHONPATH=src python -m unittest discover -v +PIP_CACHE_DIR=/tmp/pip-cache python -m pip install --root-user-action=ignore --no-deps --no-build-isolation --target . +PYTHONPATH= PROJECT_BUS_MIGRATIONS=/migrations python -m unittest discover -v + +SCOPE INTEGRITY +- Canonical Sandbox ZIP SHA-256 after implementation: + 9ca8ec98221146730c1ed82d8d5e2011b710d623ea046b7ab87a1d1cfb4e8bad +- Handoff delta SHA-256 after implementation: + 81fd1e58319299c4acfaf6c22bddacfe6f1de34ccfffac7943647d834aa912e6 +- Neither file exists inside this repository. + +PRE-VERIFICATION SOURCE SET +- Files: 27 +- Aggregate SHA-256 over sorted per-file SHA-256 records: + e379541644e82fa4fd8096761625a0ec015528e51f20b21b13c9a7fa7bf53ce1 +- This aggregate intentionally excludes this VERIFICATION.txt file. + +REMAINING GATES +1. Independent Claude security/architecture review using CLAUDE_REVIEW_BRIEF.md. +2. Correction and re-verification of any findings. +3. Exact MCP interoperability checks with intended ChatGPT and Claude Code versions. +4. PM gate acceptance. +5. Production hardening work package before internet-facing deployment. diff --git a/evidence/WP-PA-001-EVIDENCE.md b/evidence/WP-PA-001-EVIDENCE.md new file mode 100644 index 0000000..873eeef --- /dev/null +++ b/evidence/WP-PA-001-EVIDENCE.md @@ -0,0 +1,56 @@ +# WP-PA-001 Evidence + +## Scope delivered + +- standalone generic repository root; +- relational schema and migration; +- append-only project event log; +- projects, actors, formal roles, work packages, results, reviews, decisions, + escalations, links, gates, baselines, and cursor synchronization; +- authorization, identity abstraction, idempotency, atomic claims, and + concurrency-safe writes; +- MCP Streamable HTTP request/response interface; +- unit, governance, concurrency, and HTTP integration tests; +- deployment example, architecture, tool contract, threat model, production + delta, and independent review brief. + +## Verification commands + +```bash +PYTHONPATH=src python -m compileall -q src tests +PYTHONPATH=src python -m unittest discover -v +python -m unittest discover -v +``` + +The third command is run after editable installation and proves the packaged +import path. Record the final observed counts and hashes in +`VERIFICATION.txt`; generated database files are excluded. + +## Acceptance mapping + +| Requested evidence | Location | +|---|---| +| Append-only event log | `migrations/001_initial.sql`, trigger tests | +| Project/actor/role model | migration, `models.py`, service tests | +| Work/result/review lifecycle | `service.py`, full workflow test | +| Decision and gate separation | authorization tests | +| Reviewer independence | request/submit checks and test | +| Idempotency | table, canonical hash logic, replay/mismatch test | +| Concurrency safety | `BEGIN IMMEDIATE`, concurrent claim test | +| Cursor synchronization | `sync_since`, pagination test | +| Commit/artifact/baseline links | service and MCP tools | +| Authentication abstraction | `AuthProvider`, static adapter | +| MCP interface | `mcp.py`, HTTP integration tests | +| Deployment and production differences | `deploy/`, production profile | +| Threat model | `docs/THREAT_MODEL.md` | +| Independent review input | `CLAUDE_REVIEW_BRIEF.md` | + +## Explicit exclusions + +- no external repository creation, commit, or push; +- no production deployment or ChatGPT/Claude activation; +- no PostgreSQL adapter, OIDC provider, SSE push channel, or chat wake-up; +- no Sandbox/Epimonos product domain logic; +- no copy of the Sandbox SG handoff delta; +- no modification of the canonical Sandbox v0.1.0 ZIP. + diff --git a/migrations/001_initial.sql b/migrations/001_initial.sql new file mode 100644 index 0000000..a0eb758 --- /dev/null +++ b/migrations/001_initial.sql @@ -0,0 +1,144 @@ +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY, + applied_at TEXT NOT NULL +); + +CREATE TABLE projects ( + project_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL +); + +CREATE TABLE actors ( + actor_id TEXT PRIMARY KEY, + auth_subject TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL, + created_at TEXT NOT NULL +); + +CREATE TABLE project_memberships ( + project_id TEXT NOT NULL REFERENCES projects(project_id), + actor_id TEXT NOT NULL REFERENCES actors(actor_id), + role TEXT NOT NULL CHECK (role IN ( + 'SYSTEM','PO','PM','PA','ARCHITECT','LEAD_ENGINEER', + 'IMPLEMENTER','REVIEWER','SECURITY_REVIEWER','OBSERVER' + )), + active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)), + created_at TEXT NOT NULL, + PRIMARY KEY (project_id, actor_id) +); + +CREATE TABLE work_packages ( + project_id TEXT NOT NULL REFERENCES projects(project_id), + work_package_id TEXT NOT NULL, + title TEXT NOT NULL, + description TEXT NOT NULL, + acceptance_criteria_json TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ( + 'OPEN','CLAIMED','RESULT_SUBMITTED','IN_REVIEW','REVIEWED', + 'ACCEPTED','REJECTED','CANCELLED' + )), + issued_by TEXT NOT NULL REFERENCES actors(actor_id), + claimed_by TEXT REFERENCES actors(actor_id), + result_submitted_by TEXT REFERENCES actors(actor_id), + result_json TEXT, + accepted_by TEXT REFERENCES actors(actor_id), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (project_id, work_package_id) +); +CREATE INDEX work_packages_project_status + ON work_packages(project_id, status, updated_at); + +CREATE TABLE review_requests ( + project_id TEXT NOT NULL REFERENCES projects(project_id), + review_id TEXT NOT NULL, + work_package_id TEXT NOT NULL, + requested_by TEXT NOT NULL REFERENCES actors(actor_id), + reviewer_id TEXT NOT NULL REFERENCES actors(actor_id), + review_type TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('OPEN','SUBMITTED')), + verdict TEXT CHECK (verdict IN ('APPROVE','CHANGES_REQUESTED','REJECT')), + findings_json TEXT, + created_at TEXT NOT NULL, + submitted_at TEXT, + PRIMARY KEY (project_id, review_id), + FOREIGN KEY (project_id, work_package_id) + REFERENCES work_packages(project_id, work_package_id) +); +CREATE INDEX review_requests_project_status + ON review_requests(project_id, status, reviewer_id); + +CREATE TABLE decisions ( + project_id TEXT NOT NULL REFERENCES projects(project_id), + decision_id TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('PRODUCT','ARCHITECTURE','OPERATIONAL','GATE')), + title TEXT NOT NULL, + decision TEXT NOT NULL, + rationale TEXT NOT NULL, + decided_by TEXT NOT NULL REFERENCES actors(actor_id), + work_package_id TEXT, + created_at TEXT NOT NULL, + PRIMARY KEY (project_id, decision_id), + FOREIGN KEY (project_id, work_package_id) + REFERENCES work_packages(project_id, work_package_id) +); + +CREATE TABLE escalations ( + project_id TEXT NOT NULL REFERENCES projects(project_id), + escalation_id TEXT NOT NULL, + raised_by TEXT NOT NULL REFERENCES actors(actor_id), + severity TEXT NOT NULL CHECK (severity IN ('LOW','MEDIUM','HIGH','CRITICAL')), + title TEXT NOT NULL, + description TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('OPEN','RESOLVED')), + target_role TEXT, + resolution TEXT, + resolved_by TEXT REFERENCES actors(actor_id), + created_at TEXT NOT NULL, + resolved_at TEXT, + PRIMARY KEY (project_id, escalation_id) +); + +CREATE TABLE event_log ( + cursor INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + project_id TEXT NOT NULL REFERENCES projects(project_id), + event_type TEXT NOT NULL, + actor_id TEXT NOT NULL REFERENCES actors(actor_id), + actor_role TEXT NOT NULL, + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + correlation_id TEXT, + causation_event_id TEXT, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL +); +CREATE INDEX event_log_project_cursor ON event_log(project_id, cursor); +CREATE INDEX event_log_aggregate ON event_log(project_id, aggregate_type, aggregate_id, cursor); + +CREATE TRIGGER event_log_no_update +BEFORE UPDATE ON event_log +BEGIN + SELECT RAISE(ABORT, 'event_log is append-only'); +END; + +CREATE TRIGGER event_log_no_delete +BEFORE DELETE ON event_log +BEGIN + SELECT RAISE(ABORT, 'event_log is append-only'); +END; + +CREATE TABLE idempotency_records ( + project_id TEXT NOT NULL, + actor_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + operation TEXT NOT NULL, + request_hash TEXT NOT NULL, + response_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (project_id, actor_id, idempotency_key) +); diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d9999c0 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "epimonos-project-bus" +version = "0.1.0" +description = "Generic, auditable project coordination bus with an MCP Streamable HTTP interface" +readme = "README.md" +requires-python = ">=3.12" +license = { text = "MIT" } +authors = [{ name = "Epimonos contributors" }] +dependencies = [] + +[project.scripts] +project-bus = "project_bus.__main__:main" + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/src/project_bus/__init__.py b/src/project_bus/__init__.py new file mode 100644 index 0000000..e9090da --- /dev/null +++ b/src/project_bus/__init__.py @@ -0,0 +1,4 @@ +"""Epimonos Project Bus core package.""" + +__version__ = "0.1.0" + diff --git a/src/project_bus/__main__.py b/src/project_bus/__main__.py new file mode 100644 index 0000000..a2792e4 --- /dev/null +++ b/src/project_bus/__main__.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import argparse +import logging +import os + +from .auth import StaticTokenAuthProvider +from .db import Database +from .mcp import MCPApplication +from .server import ProjectBusHTTPServer, allowed_origins_from_environment +from .service import ProjectBusService + + +def parser() -> argparse.ArgumentParser: + result = argparse.ArgumentParser(description="Run the Epimonos Project Bus MCP server") + result.add_argument( + "--database", + default=os.environ.get("PROJECT_BUS_DATABASE", "project-bus.db"), + help="SQLite database path", + ) + result.add_argument("--host", default=os.environ.get("PROJECT_BUS_HOST", "127.0.0.1")) + result.add_argument( + "--port", + type=int, + default=int(os.environ.get("PROJECT_BUS_PORT", "8080")), + ) + return result + + +def main() -> None: + arguments = parser().parse_args() + logging.basicConfig( + level=os.environ.get("PROJECT_BUS_LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + service = ProjectBusService(Database(arguments.database)) + service.initialize() + application = MCPApplication(service) + auth_provider = StaticTokenAuthProvider.from_environment() + server = ProjectBusHTTPServer( + (arguments.host, arguments.port), + application, + auth_provider, + allowed_origins_from_environment(), + ) + logging.getLogger("project_bus").info( + "Project Bus listening on http://%s:%d/mcp", arguments.host, arguments.port + ) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + + +if __name__ == "__main__": + main() + diff --git a/src/project_bus/auth.py b/src/project_bus/auth.py new file mode 100644 index 0000000..87dcee2 --- /dev/null +++ b/src/project_bus/auth.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import hashlib +import hmac +import json +import os +from dataclasses import dataclass +from typing import Mapping, Protocol + +from .errors import AuthenticationError +from .models import Principal + + +class AuthProvider(Protocol): + def authenticate(self, authorization_header: str | None) -> Principal: + """Authenticate a request without trusting actor parameters.""" + + +@dataclass(frozen=True) +class TokenIdentity: + actor_id: str + auth_subject: str + display_name: str + + +class StaticTokenAuthProvider: + """Small deployment auth adapter intended for local and controlled use. + + Tokens are compared using SHA-256 digests and constant-time comparison. + Production deployments should replace this adapter with an OIDC/JWT or + mTLS-backed implementation and keep the same ``AuthProvider`` contract. + """ + + def __init__(self, token_identities: Mapping[str, TokenIdentity]): + if not token_identities: + raise ValueError("At least one authentication token is required") + if any(len(token) < 24 for token in token_identities): + raise ValueError("Every authentication token must contain at least 24 characters") + self._digests = { + hashlib.sha256(token.encode("utf-8")).digest(): identity + for token, identity in token_identities.items() + } + + @classmethod + def from_environment(cls) -> "StaticTokenAuthProvider": + raw = os.environ.get("PROJECT_BUS_TOKENS_JSON") + bootstrap = os.environ.get("PROJECT_BUS_BOOTSTRAP_TOKEN") + if not bootstrap: + raise ValueError("PROJECT_BUS_BOOTSTRAP_TOKEN is required") + if len(bootstrap) < 24: + raise ValueError("PROJECT_BUS_BOOTSTRAP_TOKEN must contain at least 24 characters") + entries: dict[str, TokenIdentity] = { + bootstrap: TokenIdentity("system", "system", "Bootstrap system") + } + if raw: + decoded = json.loads(raw) + if not isinstance(decoded, dict): + raise ValueError("PROJECT_BUS_TOKENS_JSON must be a JSON object") + for token, identity in decoded.items(): + entries[token] = TokenIdentity( + actor_id=identity["actor_id"], + auth_subject=identity.get("auth_subject", identity["actor_id"]), + display_name=identity.get("display_name", identity["actor_id"]), + ) + return cls(entries) + + def authenticate(self, authorization_header: str | None) -> Principal: + if not authorization_header or not authorization_header.startswith("Bearer "): + raise AuthenticationError("A Bearer token is required") + token = authorization_header[7:] + candidate = hashlib.sha256(token.encode("utf-8")).digest() + for expected, identity in self._digests.items(): + if hmac.compare_digest(candidate, expected): + return Principal( + actor_id=identity.actor_id, + auth_subject=identity.auth_subject, + display_name=identity.display_name, + ) + raise AuthenticationError("Invalid Bearer token") diff --git a/src/project_bus/db.py b/src/project_bus/db.py new file mode 100644 index 0000000..97a1ef9 --- /dev/null +++ b/src/project_bus/db.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import contextlib +import os +import sqlite3 +import threading +from pathlib import Path +from typing import Iterator + + +class Database: + """SQLite persistence adapter. + + Each transaction gets its own connection. ``BEGIN IMMEDIATE`` serializes + writers before domain invariants are read, preventing check-then-write + races while still allowing concurrent readers in WAL mode. + """ + + def __init__(self, path: str | Path, migrations_dir: str | Path | None = None): + self.path = str(path) + configured = migrations_dir or os.environ.get("PROJECT_BUS_MIGRATIONS") + if configured: + self.migrations_dir = Path(configured) + else: + source_tree = Path(__file__).parents[2] / "migrations" + working_tree = Path.cwd() / "migrations" + self.migrations_dir = source_tree if source_tree.is_dir() else working_tree + self._migration_lock = threading.Lock() + + def connect(self) -> sqlite3.Connection: + connection = sqlite3.connect( + self.path, + timeout=30, + isolation_level=None, + check_same_thread=False, + ) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA busy_timeout = 30000") + connection.execute("PRAGMA journal_mode = WAL") + return connection + + def migrate(self) -> None: + with self._migration_lock: + connection = self.connect() + try: + connection.execute( + "CREATE TABLE IF NOT EXISTS schema_migrations " + "(version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL)" + ) + current = { + row["version"] + for row in connection.execute("SELECT version FROM schema_migrations") + } + migration_files = sorted(self.migrations_dir.glob("[0-9][0-9][0-9]_*.sql")) + if not migration_files: + raise RuntimeError(f"No migrations found in {self.migrations_dir}") + for migration in migration_files: + version = int(migration.name.split("_", 1)[0]) + if version in current: + continue + connection.executescript(migration.read_text(encoding="utf-8")) + connection.execute( + "INSERT INTO schema_migrations(version, applied_at) " + "VALUES (?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))", + (version,), + ) + finally: + connection.close() + + @contextlib.contextmanager + def transaction(self, *, write: bool = False) -> Iterator[sqlite3.Connection]: + connection = self.connect() + try: + connection.execute("BEGIN IMMEDIATE" if write else "BEGIN") + yield connection + connection.commit() + except BaseException: + connection.rollback() + raise + finally: + connection.close() diff --git a/src/project_bus/errors.py b/src/project_bus/errors.py new file mode 100644 index 0000000..da6c66f --- /dev/null +++ b/src/project_bus/errors.py @@ -0,0 +1,30 @@ +class ProjectBusError(Exception): + """Base class for expected domain and interface failures.""" + + code = "project_bus_error" + + def __init__(self, message: str, *, details: dict | None = None): + super().__init__(message) + self.message = message + self.details = details or {} + + +class ValidationError(ProjectBusError): + code = "validation_error" + + +class NotFoundError(ProjectBusError): + code = "not_found" + + +class ConflictError(ProjectBusError): + code = "conflict" + + +class AuthorizationError(ProjectBusError): + code = "forbidden" + + +class AuthenticationError(ProjectBusError): + code = "unauthorized" + diff --git a/src/project_bus/mcp.py b/src/project_bus/mcp.py new file mode 100644 index 0000000..2c7c9fa --- /dev/null +++ b/src/project_bus/mcp.py @@ -0,0 +1,430 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Callable + +from .errors import ProjectBusError, ValidationError +from .models import Principal +from .service import ProjectBusService + + +def object_schema( + properties: dict[str, dict[str, Any]], required: list[str] +) -> dict[str, Any]: + return { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + } + + +STRING = {"type": "string"} +IDEMPOTENCY = { + "type": "string", + "description": "Caller-generated stable key, unique per actor/project mutation.", +} + + +@dataclass(frozen=True) +class Tool: + name: str + description: str + schema: dict[str, Any] + handler: Callable[..., dict[str, Any]] + + def metadata(self) -> dict[str, Any]: + return { + "name": self.name, + "description": self.description, + "inputSchema": self.schema, + } + + +class MCPApplication: + protocol_version = "2025-03-26" + + def __init__(self, service: ProjectBusService): + self.service = service + common = {"project_id": STRING} + self.tools: dict[str, Tool] = {} + + def add(name: str, description: str, properties: dict, required: list[str], handler): + self.tools[name] = Tool( + name, + description, + object_schema(properties, required), + handler, + ) + + add( + "create_project", + "Bootstrap a new isolated project. Requires the SYSTEM identity.", + { + "project_id": STRING, + "name": STRING, + "description": {"type": "string", "default": ""}, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "name", "idempotency_key"], + service.create_project, + ) + add( + "register_actor", + "Register an actor and their single formal role in a project.", + { + **common, + "actor_id": STRING, + "auth_subject": STRING, + "display_name": STRING, + "role": { + "type": "string", + "enum": [ + "PO", + "PM", + "PA", + "ARCHITECT", + "LEAD_ENGINEER", + "IMPLEMENTER", + "REVIEWER", + "SECURITY_REVIEWER", + "OBSERVER", + ], + }, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "actor_id", "auth_subject", "display_name", "role", "idempotency_key"], + service.register_actor, + ) + add( + "issue_work_package", + "Issue a formally scoped work package. Only PM or PO may issue.", + { + **common, + "work_package_id": STRING, + "title": STRING, + "description": STRING, + "acceptance_criteria": {"type": "array", "items": STRING, "minItems": 1}, + "idempotency_key": IDEMPOTENCY, + }, + [ + "project_id", + "work_package_id", + "title", + "description", + "acceptance_criteria", + "idempotency_key", + ], + service.issue_work_package, + ) + add( + "claim_work_package", + "Atomically claim an open work package as architect or implementer.", + {**common, "work_package_id": STRING, "idempotency_key": IDEMPOTENCY}, + ["project_id", "work_package_id", "idempotency_key"], + service.claim_work_package, + ) + add( + "submit_result", + "Submit implementation result and evidence for the caller's claimed work package.", + { + **common, + "work_package_id": STRING, + "summary": STRING, + "evidence": {"type": "array", "items": {"type": "object"}}, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "work_package_id", "summary", "evidence", "idempotency_key"], + service.submit_result, + ) + add( + "request_review", + "Request a review from an actor independent of the implementer.", + { + **common, + "work_package_id": STRING, + "reviewer_id": STRING, + "review_type": STRING, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "work_package_id", "reviewer_id", "review_type", "idempotency_key"], + service.request_review, + ) + add( + "submit_review", + "Submit findings and verdict for an assigned independent review.", + { + **common, + "review_id": STRING, + "verdict": { + "type": "string", + "enum": ["APPROVE", "CHANGES_REQUESTED", "REJECT"], + }, + "findings": {"type": "array", "items": {"type": "object"}}, + "summary": STRING, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "review_id", "verdict", "findings", "summary", "idempotency_key"], + service.submit_review, + ) + add( + "record_decision", + "Record an authorized, immutable product, architecture, operational, or gate decision.", + { + **common, + "decision_id": STRING, + "kind": { + "type": "string", + "enum": ["PRODUCT", "ARCHITECTURE", "OPERATIONAL", "GATE"], + }, + "title": STRING, + "decision": STRING, + "rationale": STRING, + "work_package_id": {"type": ["string", "null"], "default": None}, + "idempotency_key": IDEMPOTENCY, + }, + [ + "project_id", + "decision_id", + "kind", + "title", + "decision", + "rationale", + "idempotency_key", + ], + service.record_decision, + ) + add( + "raise_escalation", + "Raise an auditable escalation to a formal project role.", + { + **common, + "severity": { + "type": "string", + "enum": ["LOW", "MEDIUM", "HIGH", "CRITICAL"], + }, + "title": STRING, + "description": STRING, + "target_role": {"type": ["string", "null"], "default": None}, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "severity", "title", "description", "idempotency_key"], + service.raise_escalation, + ) + add( + "resolve_escalation", + "Resolve an open escalation. Only PM or PO may resolve.", + { + **common, + "escalation_id": STRING, + "resolution": STRING, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "escalation_id", "resolution", "idempotency_key"], + service.resolve_escalation, + ) + add( + "link_commit", + "Append a Git commit reference to a work package event trail.", + { + **common, + "work_package_id": STRING, + "repository": STRING, + "commit_sha": STRING, + "relation": STRING, + "idempotency_key": IDEMPOTENCY, + }, + [ + "project_id", + "work_package_id", + "repository", + "commit_sha", + "relation", + "idempotency_key", + ], + service.link_commit, + ) + add( + "link_artifact", + "Append an artifact URI and optional digest to a work package event trail.", + { + **common, + "work_package_id": STRING, + "uri": STRING, + "digest": {"type": ["string", "null"], "default": None}, + "media_type": {"type": ["string", "null"], "default": None}, + "relation": STRING, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "work_package_id", "uri", "relation", "idempotency_key"], + service.link_artifact, + ) + add( + "close_gate", + "Close a work-package gate. Only PM/PO; ACCEPT requires independent approval.", + { + **common, + "work_package_id": STRING, + "outcome": {"type": "string", "enum": ["ACCEPT", "REJECT", "HOLD"]}, + "rationale": STRING, + "idempotency_key": IDEMPOTENCY, + }, + ["project_id", "work_package_id", "outcome", "rationale", "idempotency_key"], + service.close_gate, + ) + add( + "baseline_release", + "Record a baseline release containing only accepted work packages.", + { + **common, + "version": STRING, + "artifact_uri": STRING, + "digest": STRING, + "included_work_packages": {"type": "array", "items": STRING}, + "idempotency_key": IDEMPOTENCY, + }, + [ + "project_id", + "version", + "artifact_uri", + "digest", + "included_work_packages", + "idempotency_key", + ], + service.baseline_release, + ) + add( + "sync_since", + "Read ordered immutable project events after a cursor.", + { + **common, + "cursor": {"type": "integer", "minimum": 0, "default": 0}, + "limit": {"type": "integer", "minimum": 1, "maximum": 500, "default": 100}, + }, + ["project_id"], + service.sync_since, + ) + add( + "list_pending_actions", + "List active work packages, relevant open reviews, and open escalations.", + common, + ["project_id"], + service.list_pending_actions, + ) + add( + "get_work_package", + "Read a work package, its result, and review trail.", + {**common, "work_package_id": STRING}, + ["project_id", "work_package_id"], + service.get_work_package, + ) + add( + "get_project", + "Read project metadata and its formal actor-role roster.", + common, + ["project_id"], + service.get_project, + ) + + def handle(self, request: dict[str, Any], principal: Principal | None) -> dict[str, Any] | None: + request_id = request.get("id") + method = request.get("method") + if request.get("jsonrpc") != "2.0" or not isinstance(method, str): + return self._rpc_error(request_id, -32600, "Invalid Request") + if method.startswith("notifications/"): + return None + if method == "initialize": + return self._rpc_result( + request_id, + { + "protocolVersion": self.protocol_version, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": {"name": "epimonos-project-bus", "version": "0.1.0"}, + "instructions": ( + "Use stable idempotency keys for every mutation. Actor identity is " + "derived from the authenticated transport and never from tool arguments." + ), + }, + ) + if method == "ping": + return self._rpc_result(request_id, {}) + if method == "tools/list": + return self._rpc_result( + request_id, + {"tools": [tool.metadata() for tool in self.tools.values()]}, + ) + if method == "tools/call": + if principal is None: + raise ValidationError("Authenticated principal is required") + params = request.get("params") or {} + name = params.get("name") + arguments = params.get("arguments") or {} + tool = self.tools.get(name) + if tool is None: + return self._rpc_error(request_id, -32602, f"Unknown tool: {name}") + if not isinstance(arguments, dict): + return self._rpc_error(request_id, -32602, "Tool arguments must be an object") + try: + # Optional defaults are normalized here because not all MCP clients + # materialize JSON Schema defaults. + if name == "create_project": + arguments.setdefault("description", "") + elif name == "record_decision": + arguments.setdefault("work_package_id", None) + elif name == "raise_escalation": + arguments.setdefault("target_role", None) + elif name == "link_artifact": + arguments.setdefault("digest", None) + arguments.setdefault("media_type", None) + elif name == "sync_since": + arguments.setdefault("cursor", 0) + arguments.setdefault("limit", 100) + result = tool.handler(principal, **arguments) + return self._rpc_result( + request_id, + { + "content": [ + { + "type": "text", + "text": json.dumps(result, ensure_ascii=False, sort_keys=True), + } + ], + "structuredContent": result, + "isError": False, + }, + ) + except ProjectBusError as error: + payload = { + "code": error.code, + "message": error.message, + "details": error.details, + } + return self._rpc_result( + request_id, + { + "content": [ + { + "type": "text", + "text": json.dumps(payload, ensure_ascii=False, sort_keys=True), + } + ], + "structuredContent": payload, + "isError": True, + }, + ) + except TypeError as error: + return self._rpc_error(request_id, -32602, f"Invalid tool arguments: {error}") + return self._rpc_error(request_id, -32601, "Method not found") + + @staticmethod + def _rpc_result(request_id: Any, result: dict[str, Any]) -> dict[str, Any]: + return {"jsonrpc": "2.0", "id": request_id, "result": result} + + @staticmethod + def _rpc_error(request_id: Any, code: int, message: str) -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": code, "message": message}, + } diff --git a/src/project_bus/models.py b/src/project_bus/models.py new file mode 100644 index 0000000..5f88ecf --- /dev/null +++ b/src/project_bus/models.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import Any + + +class Role(StrEnum): + SYSTEM = "SYSTEM" + PO = "PO" + PM = "PM" + PA = "PA" + ARCHITECT = "ARCHITECT" + LEAD_ENGINEER = "LEAD_ENGINEER" + IMPLEMENTER = "IMPLEMENTER" + REVIEWER = "REVIEWER" + SECURITY_REVIEWER = "SECURITY_REVIEWER" + OBSERVER = "OBSERVER" + + +class WorkPackageStatus(StrEnum): + OPEN = "OPEN" + CLAIMED = "CLAIMED" + RESULT_SUBMITTED = "RESULT_SUBMITTED" + IN_REVIEW = "IN_REVIEW" + REVIEWED = "REVIEWED" + ACCEPTED = "ACCEPTED" + REJECTED = "REJECTED" + CANCELLED = "CANCELLED" + + +class ReviewVerdict(StrEnum): + APPROVE = "APPROVE" + CHANGES_REQUESTED = "CHANGES_REQUESTED" + REJECT = "REJECT" + + +class DecisionKind(StrEnum): + PRODUCT = "PRODUCT" + ARCHITECTURE = "ARCHITECTURE" + OPERATIONAL = "OPERATIONAL" + GATE = "GATE" + + +class GateOutcome(StrEnum): + ACCEPT = "ACCEPT" + REJECT = "REJECT" + HOLD = "HOLD" + + +@dataclass(frozen=True) +class Principal: + actor_id: str + auth_subject: str + display_name: str + + +@dataclass(frozen=True) +class ActorContext: + actor_id: str + project_id: str + display_name: str + role: Role + active: bool + + +JSON = dict[str, Any] + diff --git a/src/project_bus/server.py b/src/project_bus/server.py new file mode 100644 index 0000000..ff3bc06 --- /dev/null +++ b/src/project_bus/server.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import json +import logging +import os +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from urllib.parse import urlsplit + +from .auth import AuthProvider +from .errors import AuthenticationError +from .mcp import MCPApplication + +LOGGER = logging.getLogger("project_bus.http") +MAX_BODY_BYTES = 1_048_576 + + +class ProjectBusHTTPServer(ThreadingHTTPServer): + daemon_threads = True + + def __init__( + self, + server_address: tuple[str, int], + application: MCPApplication, + auth_provider: AuthProvider, + allowed_origins: set[str], + ): + super().__init__(server_address, ProjectBusRequestHandler) + self.application = application + self.auth_provider = auth_provider + self.allowed_origins = allowed_origins + + +class ProjectBusRequestHandler(BaseHTTPRequestHandler): + server: ProjectBusHTTPServer + protocol_version = "HTTP/1.1" + + def log_message(self, format_string: str, *args: Any) -> None: + LOGGER.info("%s - %s", self.address_string(), format_string % args) + + def do_GET(self) -> None: + path = urlsplit(self.path).path + if path == "/healthz": + self._json(HTTPStatus.OK, {"status": "ok"}) + return + if path == "/mcp": + self._json( + HTTPStatus.METHOD_NOT_ALLOWED, + {"error": "This stateless server does not expose an SSE GET stream"}, + extra_headers={"Allow": "POST"}, + ) + return + self._json(HTTPStatus.NOT_FOUND, {"error": "not_found"}) + + def do_POST(self) -> None: + if urlsplit(self.path).path != "/mcp": + self._json(HTTPStatus.NOT_FOUND, {"error": "not_found"}) + return + if not self._origin_allowed(): + self._json(HTTPStatus.FORBIDDEN, {"error": "origin_not_allowed"}) + return + content_type = self.headers.get("Content-Type", "").split(";", 1)[0].strip() + if content_type != "application/json": + self._json(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, {"error": "application/json required"}) + return + try: + length = int(self.headers.get("Content-Length", "0")) + except ValueError: + self._json(HTTPStatus.BAD_REQUEST, {"error": "invalid_content_length"}) + return + if length <= 0 or length > MAX_BODY_BYTES: + self._json(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid_body_size"}) + return + try: + request = json.loads(self.rfile.read(length)) + except (json.JSONDecodeError, UnicodeDecodeError): + self._json( + HTTPStatus.OK, + {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}}, + ) + return + if not isinstance(request, dict): + self._json( + HTTPStatus.OK, + {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Invalid Request"}}, + ) + return + principal = None + if request.get("method") == "tools/call": + try: + principal = self.server.auth_provider.authenticate(self.headers.get("Authorization")) + except AuthenticationError as error: + self._json( + HTTPStatus.UNAUTHORIZED, + {"error": error.code, "message": error.message}, + extra_headers={"WWW-Authenticate": 'Bearer realm="project-bus"'}, + ) + return + response = self.server.application.handle(request, principal) + if response is None: + self.send_response(HTTPStatus.ACCEPTED) + self.send_header("Content-Length", "0") + self.end_headers() + return + self._json(HTTPStatus.OK, response) + + def _origin_allowed(self) -> bool: + origin = self.headers.get("Origin") + if not origin: + return True + return origin in self.server.allowed_origins + + def _json( + self, + status: HTTPStatus, + payload: dict[str, Any], + *, + extra_headers: dict[str, str] | None = None, + ) -> None: + body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + for name, value in (extra_headers or {}).items(): + self.send_header(name, value) + self.end_headers() + self.wfile.write(body) + + +def allowed_origins_from_environment() -> set[str]: + raw = os.environ.get("PROJECT_BUS_ALLOWED_ORIGINS", "") + return {item.strip() for item in raw.split(",") if item.strip()} + diff --git a/src/project_bus/service.py b/src/project_bus/service.py new file mode 100644 index 0000000..06ab513 --- /dev/null +++ b/src/project_bus/service.py @@ -0,0 +1,1245 @@ +from __future__ import annotations + +import hashlib +import json +import re +import sqlite3 +import uuid +from datetime import UTC, datetime +from typing import Any, Callable + +from .db import Database +from .errors import AuthorizationError, ConflictError, NotFoundError, ValidationError +from .models import ( + ActorContext, + DecisionKind, + GateOutcome, + Principal, + ReviewVerdict, + Role, + WorkPackageStatus, +) + +Mutation = Callable[[sqlite3.Connection, ActorContext], dict[str, Any]] +ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$") +COMMIT_PATTERN = re.compile(r"^[0-9a-fA-F]{7,64}$") + + +def utc_now() -> str: + return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def new_id(prefix: str) -> str: + return f"{prefix}_{uuid.uuid4().hex}" + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def as_dict(row: sqlite3.Row | None) -> dict[str, Any] | None: + return dict(row) if row is not None else None + + +class ProjectBusService: + def __init__(self, database: Database): + self.db = database + + def initialize(self) -> None: + self.db.migrate() + now = utc_now() + with self.db.transaction(write=True) as connection: + connection.execute( + "INSERT OR IGNORE INTO actors(actor_id, auth_subject, display_name, created_at) " + "VALUES ('system', 'system', 'Bootstrap system', ?)", + (now,), + ) + + @staticmethod + def _validate_id(name: str, value: str) -> str: + if not isinstance(value, str) or not ID_PATTERN.fullmatch(value): + raise ValidationError( + f"{name} must be 1-128 safe identifier characters", + details={"field": name}, + ) + return value + + @staticmethod + def _require_text(name: str, value: str, maximum: int = 20_000) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValidationError(f"{name} must not be empty", details={"field": name}) + if len(value) > maximum: + raise ValidationError(f"{name} exceeds {maximum} characters", details={"field": name}) + return value.strip() + + @staticmethod + def _require_roles(context: ActorContext, *roles: Role) -> None: + if context.role not in roles: + raise AuthorizationError( + f"Role {context.role} may not perform this operation", + details={"required_roles": [role.value for role in roles]}, + ) + + @staticmethod + def _actor_context( + connection: sqlite3.Connection, project_id: str, principal: Principal + ) -> ActorContext: + row = connection.execute( + "SELECT a.actor_id, a.display_name, m.project_id, m.role, m.active " + "FROM actors a JOIN project_memberships m ON m.actor_id = a.actor_id " + "WHERE a.actor_id = ? AND a.auth_subject = ? AND m.project_id = ?", + (principal.actor_id, principal.auth_subject, project_id), + ).fetchone() + if row is None: + raise AuthorizationError("Authenticated actor is not a member of this project") + context = ActorContext( + actor_id=row["actor_id"], + project_id=row["project_id"], + display_name=row["display_name"], + role=Role(row["role"]), + active=bool(row["active"]), + ) + if not context.active: + raise AuthorizationError("Project membership is inactive") + return context + + @staticmethod + def _event( + connection: sqlite3.Connection, + *, + project_id: str, + context: ActorContext, + event_type: str, + aggregate_type: str, + aggregate_id: str, + payload: dict[str, Any], + correlation_id: str | None = None, + causation_event_id: str | None = None, + ) -> dict[str, Any]: + event_id = new_id("evt") + created_at = utc_now() + cursor = connection.execute( + "INSERT INTO event_log(" + "event_id, project_id, event_type, actor_id, actor_role, aggregate_type, " + "aggregate_id, correlation_id, causation_event_id, payload_json, created_at" + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + event_id, + project_id, + event_type, + context.actor_id, + context.role.value, + aggregate_type, + aggregate_id, + correlation_id, + causation_event_id, + canonical_json(payload), + created_at, + ), + ).lastrowid + return { + "event_id": event_id, + "cursor": cursor, + "event_type": event_type, + "created_at": created_at, + } + + def _mutate( + self, + *, + project_id: str, + principal: Principal, + operation: str, + idempotency_key: str, + request: dict[str, Any], + callback: Mutation, + ) -> dict[str, Any]: + self._validate_id("project_id", project_id) + self._validate_id("idempotency_key", idempotency_key) + request_hash = hashlib.sha256(canonical_json(request).encode("utf-8")).hexdigest() + with self.db.transaction(write=True) as connection: + context = self._actor_context(connection, project_id, principal) + prior = connection.execute( + "SELECT operation, request_hash, response_json FROM idempotency_records " + "WHERE project_id = ? AND actor_id = ? AND idempotency_key = ?", + (project_id, context.actor_id, idempotency_key), + ).fetchone() + if prior: + if prior["operation"] != operation or prior["request_hash"] != request_hash: + raise ConflictError( + "Idempotency key was already used with a different request", + details={"idempotency_key": idempotency_key}, + ) + response = json.loads(prior["response_json"]) + response["idempotent_replay"] = True + return response + response = callback(connection, context) + response["idempotent_replay"] = False + connection.execute( + "INSERT INTO idempotency_records(" + "project_id, actor_id, idempotency_key, operation, request_hash, " + "response_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + project_id, + context.actor_id, + idempotency_key, + operation, + request_hash, + canonical_json(response), + utc_now(), + ), + ) + return response + + def create_project( + self, + principal: Principal, + *, + project_id: str, + name: str, + description: str = "", + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("project_id", project_id) + self._validate_id("idempotency_key", idempotency_key) + name = self._require_text("name", name, 200) + if principal.actor_id != "system" or principal.auth_subject != "system": + raise AuthorizationError("Only the bootstrap system may create a project") + request = {"project_id": project_id, "name": name, "description": description} + request_hash = hashlib.sha256(canonical_json(request).encode("utf-8")).hexdigest() + with self.db.transaction(write=True) as connection: + exists = connection.execute( + "SELECT 1 FROM projects WHERE project_id = ?", (project_id,) + ).fetchone() + if exists: + prior = connection.execute( + "SELECT operation, request_hash, response_json FROM idempotency_records " + "WHERE project_id = ? AND actor_id = 'system' AND idempotency_key = ?", + (project_id, idempotency_key), + ).fetchone() + if prior and prior["operation"] == "create_project" and prior["request_hash"] == request_hash: + response = json.loads(prior["response_json"]) + response["idempotent_replay"] = True + return response + raise ConflictError("Project already exists") + now = utc_now() + connection.execute( + "INSERT INTO projects(project_id, name, description, created_at) VALUES (?, ?, ?, ?)", + (project_id, name, description, now), + ) + connection.execute( + "INSERT INTO project_memberships(project_id, actor_id, role, active, created_at) " + "VALUES (?, 'system', 'SYSTEM', 1, ?)", + (project_id, now), + ) + context = ActorContext("system", project_id, "Bootstrap system", Role.SYSTEM, True) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="project.created", + aggregate_type="project", + aggregate_id=project_id, + payload={"name": name, "description": description}, + ) + response = {"project_id": project_id, "event": event, "idempotent_replay": False} + connection.execute( + "INSERT INTO idempotency_records VALUES (?, 'system', ?, 'create_project', ?, ?, ?)", + (project_id, idempotency_key, request_hash, canonical_json(response), now), + ) + return response + + def register_actor( + self, + principal: Principal, + *, + project_id: str, + actor_id: str, + auth_subject: str, + display_name: str, + role: str, + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("actor_id", actor_id) + self._validate_id("auth_subject", auth_subject) + display_name = self._require_text("display_name", display_name, 200) + try: + requested_role = Role(role) + except ValueError as error: + raise ValidationError("Unknown role", details={"role": role}) from error + if requested_role == Role.SYSTEM: + raise AuthorizationError("SYSTEM memberships cannot be delegated") + request = { + "actor_id": actor_id, + "auth_subject": auth_subject, + "display_name": display_name, + "role": requested_role.value, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.SYSTEM, Role.PO, Role.PM) + if context.role == Role.PM and requested_role == Role.PO: + raise AuthorizationError("PM may not appoint a PO") + existing = connection.execute( + "SELECT actor_id, auth_subject FROM actors WHERE actor_id = ? OR auth_subject = ?", + (actor_id, auth_subject), + ).fetchall() + if existing and any( + row["actor_id"] != actor_id or row["auth_subject"] != auth_subject for row in existing + ): + raise ConflictError("actor_id or auth_subject is already assigned") + now = utc_now() + connection.execute( + "INSERT OR IGNORE INTO actors(actor_id, auth_subject, display_name, created_at) " + "VALUES (?, ?, ?, ?)", + (actor_id, auth_subject, display_name, now), + ) + membership = connection.execute( + "SELECT role FROM project_memberships WHERE project_id = ? AND actor_id = ?", + (project_id, actor_id), + ).fetchone() + if membership: + raise ConflictError("Actor is already a project member") + connection.execute( + "INSERT INTO project_memberships(project_id, actor_id, role, active, created_at) " + "VALUES (?, ?, ?, 1, ?)", + (project_id, actor_id, requested_role.value, now), + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="actor.registered", + aggregate_type="actor", + aggregate_id=actor_id, + payload={"auth_subject": auth_subject, "display_name": display_name, "role": role}, + ) + return {"actor_id": actor_id, "role": role, "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="register_actor", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def issue_work_package( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + title: str, + description: str, + acceptance_criteria: list[str], + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("work_package_id", work_package_id) + title = self._require_text("title", title, 300) + description = self._require_text("description", description) + if not acceptance_criteria or not all(isinstance(item, str) and item.strip() for item in acceptance_criteria): + raise ValidationError("acceptance_criteria must contain non-empty strings") + request = { + "work_package_id": work_package_id, + "title": title, + "description": description, + "acceptance_criteria": acceptance_criteria, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.PO, Role.PM) + now = utc_now() + try: + connection.execute( + "INSERT INTO work_packages(" + "work_package_id, project_id, title, description, acceptance_criteria_json, " + "status, issued_by, created_at, updated_at" + ") VALUES (?, ?, ?, ?, ?, 'OPEN', ?, ?, ?)", + ( + work_package_id, + project_id, + title, + description, + canonical_json(acceptance_criteria), + context.actor_id, + now, + now, + ), + ) + except sqlite3.IntegrityError as error: + raise ConflictError("Work package already exists") from error + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="work_package.issued", + aggregate_type="work_package", + aggregate_id=work_package_id, + payload=request, + correlation_id=work_package_id, + ) + return {"work_package_id": work_package_id, "status": "OPEN", "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="issue_work_package", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def claim_work_package( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("work_package_id", work_package_id) + request = {"work_package_id": work_package_id} + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.ARCHITECT, Role.LEAD_ENGINEER, Role.IMPLEMENTER) + result = connection.execute( + "UPDATE work_packages SET status = 'CLAIMED', claimed_by = ?, updated_at = ? " + "WHERE project_id = ? AND work_package_id = ? AND status = 'OPEN'", + (context.actor_id, utc_now(), project_id, work_package_id), + ) + if result.rowcount != 1: + row = connection.execute( + "SELECT status FROM work_packages WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone() + if row is None: + raise NotFoundError("Work package not found") + raise ConflictError("Work package is not open", details={"status": row["status"]}) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="work_package.claimed", + aggregate_type="work_package", + aggregate_id=work_package_id, + payload={"claimed_by": context.actor_id}, + correlation_id=work_package_id, + ) + return {"work_package_id": work_package_id, "status": "CLAIMED", "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="claim_work_package", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def submit_result( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + summary: str, + evidence: list[dict[str, Any]], + idempotency_key: str, + ) -> dict[str, Any]: + summary = self._require_text("summary", summary) + if not isinstance(evidence, list): + raise ValidationError("evidence must be a list") + request = {"work_package_id": work_package_id, "summary": summary, "evidence": evidence} + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + row = connection.execute( + "SELECT status, claimed_by FROM work_packages " + "WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone() + if row is None: + raise NotFoundError("Work package not found") + if row["status"] != WorkPackageStatus.CLAIMED: + raise ConflictError("Only a claimed work package can receive a result") + if row["claimed_by"] != context.actor_id: + raise AuthorizationError("Only the claiming actor may submit the result") + now = utc_now() + connection.execute( + "UPDATE work_packages SET status = 'RESULT_SUBMITTED', result_submitted_by = ?, " + "result_json = ?, updated_at = ? " + "WHERE project_id = ? AND work_package_id = ?", + ( + context.actor_id, + canonical_json({"summary": summary, "evidence": evidence}), + now, + project_id, + work_package_id, + ), + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="work_package.result_submitted", + aggregate_type="work_package", + aggregate_id=work_package_id, + payload={"summary": summary, "evidence": evidence}, + correlation_id=work_package_id, + ) + return {"work_package_id": work_package_id, "status": "RESULT_SUBMITTED", "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="submit_result", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def request_review( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + reviewer_id: str, + review_type: str, + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("reviewer_id", reviewer_id) + review_type = self._require_text("review_type", review_type, 100) + request = { + "work_package_id": work_package_id, + "reviewer_id": reviewer_id, + "review_type": review_type, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles( + context, Role.PO, Role.PM, Role.PA, Role.LEAD_ENGINEER, Role.IMPLEMENTER, Role.ARCHITECT + ) + work_package = connection.execute( + "SELECT status, claimed_by, result_submitted_by FROM work_packages " + "WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone() + if work_package is None: + raise NotFoundError("Work package not found") + if work_package["status"] not in ( + WorkPackageStatus.RESULT_SUBMITTED, + WorkPackageStatus.REVIEWED, + ): + raise ConflictError("A submitted result is required before review") + reviewer = connection.execute( + "SELECT role, active FROM project_memberships WHERE project_id = ? AND actor_id = ?", + (project_id, reviewer_id), + ).fetchone() + if reviewer is None or not reviewer["active"]: + raise ValidationError("Reviewer is not an active project member") + if Role(reviewer["role"]) not in ( + Role.REVIEWER, + Role.SECURITY_REVIEWER, + Role.ARCHITECT, + ): + raise ValidationError("Selected actor does not hold an independent review role") + if reviewer_id in (work_package["claimed_by"], work_package["result_submitted_by"]): + raise AuthorizationError("Implementer and reviewer must be independent actors") + review_id = new_id("rev") + now = utc_now() + connection.execute( + "INSERT INTO review_requests(" + "review_id, project_id, work_package_id, requested_by, reviewer_id, " + "review_type, status, created_at) VALUES (?, ?, ?, ?, ?, ?, 'OPEN', ?)", + ( + review_id, + project_id, + work_package_id, + context.actor_id, + reviewer_id, + review_type, + now, + ), + ) + connection.execute( + "UPDATE work_packages SET status = 'IN_REVIEW', updated_at = ? " + "WHERE project_id = ? AND work_package_id = ?", + (now, project_id, work_package_id), + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="review.requested", + aggregate_type="review", + aggregate_id=review_id, + payload=request, + correlation_id=work_package_id, + ) + return {"review_id": review_id, "work_package_id": work_package_id, "status": "OPEN", "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="request_review", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def submit_review( + self, + principal: Principal, + *, + project_id: str, + review_id: str, + verdict: str, + findings: list[dict[str, Any]], + summary: str, + idempotency_key: str, + ) -> dict[str, Any]: + try: + review_verdict = ReviewVerdict(verdict) + except ValueError as error: + raise ValidationError("Unknown review verdict") from error + summary = self._require_text("summary", summary) + request = { + "review_id": review_id, + "verdict": review_verdict.value, + "findings": findings, + "summary": summary, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.REVIEWER, Role.SECURITY_REVIEWER, Role.ARCHITECT) + review = connection.execute( + "SELECT r.*, w.claimed_by, w.result_submitted_by " + "FROM review_requests r JOIN work_packages w " + "ON w.project_id = r.project_id AND w.work_package_id = r.work_package_id " + "WHERE r.project_id = ? AND r.review_id = ?", + (project_id, review_id), + ).fetchone() + if review is None: + raise NotFoundError("Review request not found") + if review["status"] != "OPEN": + raise ConflictError("Review was already submitted") + if review["reviewer_id"] != context.actor_id: + raise AuthorizationError("Only the assigned reviewer may submit this review") + if context.actor_id in (review["claimed_by"], review["result_submitted_by"]): + raise AuthorizationError("Implementer may not review their own result") + now = utc_now() + connection.execute( + "UPDATE review_requests SET status = 'SUBMITTED', verdict = ?, findings_json = ?, " + "submitted_at = ? WHERE project_id = ? AND review_id = ?", + ( + review_verdict.value, + canonical_json({"summary": summary, "findings": findings}), + now, + project_id, + review_id, + ), + ) + connection.execute( + "UPDATE work_packages SET status = 'REVIEWED', updated_at = ? " + "WHERE project_id = ? AND work_package_id = ?", + (now, project_id, review["work_package_id"]), + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="review.submitted", + aggregate_type="review", + aggregate_id=review_id, + payload=request, + correlation_id=review["work_package_id"], + ) + return { + "review_id": review_id, + "work_package_id": review["work_package_id"], + "verdict": review_verdict.value, + "event": event, + } + + return self._mutate( + project_id=project_id, + principal=principal, + operation="submit_review", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def record_decision( + self, + principal: Principal, + *, + project_id: str, + decision_id: str, + kind: str, + title: str, + decision: str, + rationale: str, + work_package_id: str | None, + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("decision_id", decision_id) + try: + decision_kind = DecisionKind(kind) + except ValueError as error: + raise ValidationError("Unknown decision kind") from error + title = self._require_text("title", title, 300) + decision = self._require_text("decision", decision) + rationale = self._require_text("rationale", rationale) + request = { + "decision_id": decision_id, + "kind": decision_kind.value, + "title": title, + "decision": decision, + "rationale": rationale, + "work_package_id": work_package_id, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + if context.role == Role.PA: + raise AuthorizationError("PA has no decision mandate") + if decision_kind == DecisionKind.PRODUCT: + self._require_roles(context, Role.PO) + else: + self._require_roles(context, Role.PO, Role.PM) + if work_package_id and not connection.execute( + "SELECT 1 FROM work_packages WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone(): + raise NotFoundError("Referenced work package not found") + try: + connection.execute( + "INSERT INTO decisions(" + "decision_id, project_id, kind, title, decision, rationale, decided_by, " + "work_package_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + decision_id, + project_id, + decision_kind.value, + title, + decision, + rationale, + context.actor_id, + work_package_id, + utc_now(), + ), + ) + except sqlite3.IntegrityError as error: + raise ConflictError("Decision already exists") from error + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="decision.recorded", + aggregate_type="decision", + aggregate_id=decision_id, + payload=request, + correlation_id=work_package_id, + ) + return {"decision_id": decision_id, "kind": decision_kind.value, "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="record_decision", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def raise_escalation( + self, + principal: Principal, + *, + project_id: str, + severity: str, + title: str, + description: str, + target_role: str | None, + idempotency_key: str, + ) -> dict[str, Any]: + if severity not in ("LOW", "MEDIUM", "HIGH", "CRITICAL"): + raise ValidationError("Unknown escalation severity") + title = self._require_text("title", title, 300) + description = self._require_text("description", description) + if target_role: + try: + Role(target_role) + except ValueError as error: + raise ValidationError("Unknown target role") from error + request = { + "severity": severity, + "title": title, + "description": description, + "target_role": target_role, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + escalation_id = new_id("esc") + connection.execute( + "INSERT INTO escalations(" + "escalation_id, project_id, raised_by, severity, title, description, status, " + "target_role, created_at) VALUES (?, ?, ?, ?, ?, ?, 'OPEN', ?, ?)", + ( + escalation_id, + project_id, + context.actor_id, + severity, + title, + description, + target_role, + utc_now(), + ), + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="escalation.raised", + aggregate_type="escalation", + aggregate_id=escalation_id, + payload=request, + ) + return {"escalation_id": escalation_id, "status": "OPEN", "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="raise_escalation", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def resolve_escalation( + self, + principal: Principal, + *, + project_id: str, + escalation_id: str, + resolution: str, + idempotency_key: str, + ) -> dict[str, Any]: + self._validate_id("escalation_id", escalation_id) + resolution = self._require_text("resolution", resolution) + request = {"escalation_id": escalation_id, "resolution": resolution} + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.PO, Role.PM) + result = connection.execute( + "UPDATE escalations SET status = 'RESOLVED', resolution = ?, " + "resolved_by = ?, resolved_at = ? " + "WHERE project_id = ? AND escalation_id = ? AND status = 'OPEN'", + (resolution, context.actor_id, utc_now(), project_id, escalation_id), + ) + if result.rowcount != 1: + row = connection.execute( + "SELECT status FROM escalations WHERE project_id = ? AND escalation_id = ?", + (project_id, escalation_id), + ).fetchone() + if row is None: + raise NotFoundError("Escalation not found") + raise ConflictError("Escalation is already resolved") + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="escalation.resolved", + aggregate_type="escalation", + aggregate_id=escalation_id, + payload=request, + ) + return {"escalation_id": escalation_id, "status": "RESOLVED", "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="resolve_escalation", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def link_commit( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + repository: str, + commit_sha: str, + relation: str, + idempotency_key: str, + ) -> dict[str, Any]: + repository = self._require_text("repository", repository, 500) + if not COMMIT_PATTERN.fullmatch(commit_sha): + raise ValidationError("commit_sha must be a hexadecimal Git object id") + relation = self._require_text("relation", relation, 100) + payload = { + "work_package_id": work_package_id, + "repository": repository, + "commit_sha": commit_sha.lower(), + "relation": relation, + } + return self._link( + principal, + project_id=project_id, + work_package_id=work_package_id, + event_type="commit.linked", + payload=payload, + idempotency_key=idempotency_key, + ) + + def link_artifact( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + uri: str, + digest: str | None, + media_type: str | None, + relation: str, + idempotency_key: str, + ) -> dict[str, Any]: + uri = self._require_text("uri", uri, 2_000) + relation = self._require_text("relation", relation, 100) + payload = { + "work_package_id": work_package_id, + "uri": uri, + "digest": digest, + "media_type": media_type, + "relation": relation, + } + return self._link( + principal, + project_id=project_id, + work_package_id=work_package_id, + event_type="artifact.linked", + payload=payload, + idempotency_key=idempotency_key, + ) + + def _link( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + event_type: str, + payload: dict[str, Any], + idempotency_key: str, + ) -> dict[str, Any]: + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + if not connection.execute( + "SELECT 1 FROM work_packages WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone(): + raise NotFoundError("Work package not found") + event = self._event( + connection, + project_id=project_id, + context=context, + event_type=event_type, + aggregate_type="work_package", + aggregate_id=work_package_id, + payload=payload, + correlation_id=work_package_id, + ) + return {"work_package_id": work_package_id, "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation=event_type, + idempotency_key=idempotency_key, + request=payload, + callback=action, + ) + + def close_gate( + self, + principal: Principal, + *, + project_id: str, + work_package_id: str, + outcome: str, + rationale: str, + idempotency_key: str, + ) -> dict[str, Any]: + try: + gate_outcome = GateOutcome(outcome) + except ValueError as error: + raise ValidationError("Unknown gate outcome") from error + rationale = self._require_text("rationale", rationale) + request = { + "work_package_id": work_package_id, + "outcome": gate_outcome.value, + "rationale": rationale, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.PO, Role.PM) + work_package = connection.execute( + "SELECT status, result_submitted_by FROM work_packages " + "WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone() + if work_package is None: + raise NotFoundError("Work package not found") + if context.actor_id == work_package["result_submitted_by"]: + raise AuthorizationError("Implementer may not accept their own result") + if work_package["status"] in ( + WorkPackageStatus.ACCEPTED, + WorkPackageStatus.REJECTED, + WorkPackageStatus.CANCELLED, + ): + raise ConflictError( + "A terminal work-package gate cannot be closed again", + details={"status": work_package["status"]}, + ) + if gate_outcome == GateOutcome.ACCEPT: + if work_package["status"] != WorkPackageStatus.REVIEWED: + raise ConflictError("Only a reviewed work package can be accepted") + latest_review = connection.execute( + "SELECT verdict FROM review_requests " + "WHERE project_id = ? AND work_package_id = ? AND status = 'SUBMITTED' " + "ORDER BY submitted_at DESC, review_id DESC LIMIT 1", + (project_id, work_package_id), + ).fetchone() + if latest_review is None or latest_review["verdict"] != ReviewVerdict.APPROVE: + raise ConflictError("The latest independent review must approve") + new_status = WorkPackageStatus.ACCEPTED + elif gate_outcome == GateOutcome.REJECT: + new_status = WorkPackageStatus.REJECTED + else: + new_status = WorkPackageStatus(work_package["status"]) + connection.execute( + "UPDATE work_packages SET status = ?, accepted_by = ?, updated_at = ? " + "WHERE project_id = ? AND work_package_id = ?", + ( + new_status.value, + context.actor_id if gate_outcome == GateOutcome.ACCEPT else None, + utc_now(), + project_id, + work_package_id, + ), + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="gate.closed", + aggregate_type="work_package", + aggregate_id=work_package_id, + payload=request, + correlation_id=work_package_id, + ) + return {"work_package_id": work_package_id, "status": new_status.value, "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="close_gate", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def baseline_release( + self, + principal: Principal, + *, + project_id: str, + version: str, + artifact_uri: str, + digest: str, + included_work_packages: list[str], + idempotency_key: str, + ) -> dict[str, Any]: + version = self._require_text("version", version, 100) + artifact_uri = self._require_text("artifact_uri", artifact_uri, 2_000) + digest = self._require_text("digest", digest, 200) + request = { + "version": version, + "artifact_uri": artifact_uri, + "digest": digest, + "included_work_packages": included_work_packages, + } + + def action(connection: sqlite3.Connection, context: ActorContext) -> dict[str, Any]: + self._require_roles(context, Role.PO, Role.PM) + if connection.execute( + "SELECT 1 FROM event_log WHERE project_id = ? " + "AND event_type = 'baseline.released' AND aggregate_id = ?", + (project_id, version), + ).fetchone(): + raise ConflictError("Baseline version was already released") + if included_work_packages: + placeholders = ",".join("?" for _ in included_work_packages) + rows = connection.execute( + f"SELECT work_package_id, status FROM work_packages " + f"WHERE project_id = ? AND work_package_id IN ({placeholders})", + (project_id, *included_work_packages), + ).fetchall() + states = {row["work_package_id"]: row["status"] for row in rows} + missing_or_unaccepted = [ + item for item in included_work_packages if states.get(item) != "ACCEPTED" + ] + if missing_or_unaccepted: + raise ConflictError( + "Baseline may include only accepted work packages", + details={"not_accepted": missing_or_unaccepted}, + ) + event = self._event( + connection, + project_id=project_id, + context=context, + event_type="baseline.released", + aggregate_type="baseline", + aggregate_id=version, + payload=request, + ) + return {"version": version, "event": event} + + return self._mutate( + project_id=project_id, + principal=principal, + operation="baseline_release", + idempotency_key=idempotency_key, + request=request, + callback=action, + ) + + def sync_since( + self, + principal: Principal, + *, + project_id: str, + cursor: int = 0, + limit: int = 100, + ) -> dict[str, Any]: + if cursor < 0: + raise ValidationError("cursor must be non-negative") + if limit < 1 or limit > 500: + raise ValidationError("limit must be between 1 and 500") + with self.db.transaction() as connection: + self._actor_context(connection, project_id, principal) + rows = connection.execute( + "SELECT * FROM event_log WHERE project_id = ? AND cursor > ? " + "ORDER BY cursor ASC LIMIT ?", + (project_id, cursor, limit + 1), + ).fetchall() + has_more = len(rows) > limit + rows = rows[:limit] + events = [] + for row in rows: + event = dict(row) + event["payload"] = json.loads(event.pop("payload_json")) + events.append(event) + next_cursor = events[-1]["cursor"] if events else cursor + return {"events": events, "next_cursor": next_cursor, "has_more": has_more} + + def list_pending_actions( + self, principal: Principal, *, project_id: str + ) -> dict[str, Any]: + with self.db.transaction() as connection: + context = self._actor_context(connection, project_id, principal) + work_packages = [ + dict(row) + for row in connection.execute( + "SELECT work_package_id, title, status, claimed_by, updated_at " + "FROM work_packages WHERE project_id = ? " + "AND status NOT IN ('ACCEPTED','REJECTED','CANCELLED') ORDER BY updated_at", + (project_id,), + ) + ] + reviews = [ + dict(row) + for row in connection.execute( + "SELECT review_id, work_package_id, reviewer_id, review_type, created_at " + "FROM review_requests WHERE project_id = ? AND status = 'OPEN' " + "AND (? IN ('PO','PM','PA') OR reviewer_id = ?) ORDER BY created_at", + (project_id, context.role.value, context.actor_id), + ) + ] + escalations = [ + dict(row) + for row in connection.execute( + "SELECT escalation_id, severity, title, target_role, created_at " + "FROM escalations WHERE project_id = ? AND status = 'OPEN' " + "AND (target_role IS NULL OR target_role = ? OR ? IN ('PO','PM','PA')) " + "ORDER BY created_at", + (project_id, context.role.value, context.role.value), + ) + ] + return { + "work_packages": work_packages, + "reviews": reviews, + "escalations": escalations, + } + + def get_work_package( + self, principal: Principal, *, project_id: str, work_package_id: str + ) -> dict[str, Any]: + with self.db.transaction() as connection: + self._actor_context(connection, project_id, principal) + row = connection.execute( + "SELECT * FROM work_packages WHERE project_id = ? AND work_package_id = ?", + (project_id, work_package_id), + ).fetchone() + if row is None: + raise NotFoundError("Work package not found") + result = dict(row) + result["acceptance_criteria"] = json.loads(result.pop("acceptance_criteria_json")) + if result["result_json"]: + result["result"] = json.loads(result.pop("result_json")) + else: + result.pop("result_json") + result["result"] = None + result["reviews"] = [ + self._decode_review(review) + for review in connection.execute( + "SELECT review_id, requested_by, reviewer_id, review_type, status, verdict, " + "findings_json, created_at, submitted_at FROM review_requests " + "WHERE project_id = ? AND work_package_id = ? ORDER BY created_at", + (project_id, work_package_id), + ) + ] + return result + + def get_project(self, principal: Principal, *, project_id: str) -> dict[str, Any]: + with self.db.transaction() as connection: + self._actor_context(connection, project_id, principal) + project = connection.execute( + "SELECT project_id, name, description, created_at FROM projects " + "WHERE project_id = ?", + (project_id,), + ).fetchone() + if project is None: + raise NotFoundError("Project not found") + result = dict(project) + result["actors"] = [ + dict(row) + for row in connection.execute( + "SELECT a.actor_id, a.display_name, m.role, m.active, m.created_at " + "FROM project_memberships m JOIN actors a ON a.actor_id = m.actor_id " + "WHERE m.project_id = ? ORDER BY m.created_at, a.actor_id", + (project_id,), + ) + ] + return result + + @staticmethod + def _decode_review(review: sqlite3.Row) -> dict[str, Any]: + result = dict(review) + raw = result.pop("findings_json") + result["review"] = json.loads(raw) if raw else None + return result diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/test_mcp_http.py b/tests/test_mcp_http.py new file mode 100644 index 0000000..530b7b0 --- /dev/null +++ b/tests/test_mcp_http.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import json +import tempfile +import threading +import unittest +import urllib.error +import urllib.request +from pathlib import Path + +from project_bus.auth import StaticTokenAuthProvider, TokenIdentity +from project_bus.db import Database +from project_bus.mcp import MCPApplication +from project_bus.server import ProjectBusHTTPServer +from project_bus.service import ProjectBusService + + +class MCPHTTPTestCase(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + service = ProjectBusService(Database(Path(self.temporary.name) / "bus.db")) + service.initialize() + auth = StaticTokenAuthProvider( + {"bootstrap-secret-at-least-24": TokenIdentity("system", "system", "System")} + ) + self.server = ProjectBusHTTPServer( + ("127.0.0.1", 0), MCPApplication(service), auth, {"https://chat.example"} + ) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + self.url = f"http://127.0.0.1:{self.server.server_port}/mcp" + + def tearDown(self) -> None: + self.server.shutdown() + self.server.server_close() + self.thread.join() + self.temporary.cleanup() + + def request( + self, + payload: dict, + *, + token: str | None = None, + origin: str | None = None, + ) -> tuple[int, dict]: + headers = {"Content-Type": "application/json"} + if token: + headers["Authorization"] = f"Bearer {token}" + if origin: + headers["Origin"] = origin + request = urllib.request.Request( + self.url, + data=json.dumps(payload).encode(), + headers=headers, + method="POST", + ) + try: + response = urllib.request.urlopen(request, timeout=2) + return response.status, json.loads(response.read()) + except urllib.error.HTTPError as error: + return error.code, json.loads(error.read()) + + def test_initialize_and_list_tools(self) -> None: + status, initialized = self.request( + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}} + ) + self.assertEqual(status, 200) + self.assertEqual(initialized["result"]["protocolVersion"], "2025-03-26") + status, listed = self.request( + {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}} + ) + self.assertEqual(status, 200) + names = {tool["name"] for tool in listed["result"]["tools"]} + self.assertIn("sync_since", names) + self.assertIn("close_gate", names) + + def test_authenticated_tool_call(self) -> None: + status, response = self.request( + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "create_project", + "arguments": { + "project_id": "demo", + "name": "Demo", + "idempotency_key": "create-demo", + }, + }, + }, + token="bootstrap-secret-at-least-24", + ) + self.assertEqual(status, 200) + self.assertFalse(response["result"]["isError"]) + self.assertEqual(response["result"]["structuredContent"]["project_id"], "demo") + + def test_tool_call_requires_authentication(self) -> None: + status, response = self.request( + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": "sync_since", "arguments": {"project_id": "demo"}}, + } + ) + self.assertEqual(status, 401) + self.assertEqual(response["error"], "unauthorized") + + def test_origin_allowlist(self) -> None: + status, response = self.request( + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}, + origin="https://evil.example", + ) + self.assertEqual(status, 403) + self.assertEqual(response["error"], "origin_not_allowed") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..82369f9 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,428 @@ +from __future__ import annotations + +import sqlite3 +import tempfile +import threading +import unittest +from pathlib import Path + +from project_bus.db import Database +from project_bus.errors import AuthorizationError, ConflictError, ValidationError +from project_bus.models import Principal +from project_bus.service import ProjectBusService + + +class ServiceTestCase(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.database_path = Path(self.temporary.name) / "bus.db" + self.database = Database(self.database_path) + self.service = ProjectBusService(self.database) + self.service.initialize() + self.system = Principal("system", "system", "System") + self.service.create_project( + self.system, + project_id="sandbox", + name="Sandbox", + idempotency_key="create-sandbox", + ) + self.principals: dict[str, Principal] = {} + for actor_id, role in ( + ("po", "PO"), + ("pm", "PM"), + ("pa", "PA"), + ("lead", "LEAD_ENGINEER"), + ("impl2", "IMPLEMENTER"), + ("reviewer", "REVIEWER"), + ("security", "SECURITY_REVIEWER"), + ): + self.service.register_actor( + self.system, + project_id="sandbox", + actor_id=actor_id, + auth_subject=f"subject-{actor_id}", + display_name=actor_id.title(), + role=role, + idempotency_key=f"register-{actor_id}", + ) + self.principals[actor_id] = Principal( + actor_id, f"subject-{actor_id}", actor_id.title() + ) + + def tearDown(self) -> None: + self.temporary.cleanup() + + def issue(self, work_package_id: str = "WP-001") -> None: + self.service.issue_work_package( + self.principals["pm"], + project_id="sandbox", + work_package_id=work_package_id, + title="Build it", + description="Implement the bounded feature.", + acceptance_criteria=["Tests pass", "Evidence linked"], + idempotency_key=f"issue-{work_package_id}", + ) + + def submit_and_review(self, work_package_id: str = "WP-001") -> str: + self.service.claim_work_package( + self.principals["lead"], + project_id="sandbox", + work_package_id=work_package_id, + idempotency_key=f"claim-{work_package_id}", + ) + self.service.submit_result( + self.principals["lead"], + project_id="sandbox", + work_package_id=work_package_id, + summary="Implemented.", + evidence=[{"type": "test", "value": "all green"}], + idempotency_key=f"result-{work_package_id}", + ) + requested = self.service.request_review( + self.principals["pa"], + project_id="sandbox", + work_package_id=work_package_id, + reviewer_id="reviewer", + review_type="technical", + idempotency_key=f"request-review-{work_package_id}", + ) + self.service.submit_review( + self.principals["reviewer"], + project_id="sandbox", + review_id=requested["review_id"], + verdict="APPROVE", + findings=[], + summary="Approved independently.", + idempotency_key=f"review-{work_package_id}", + ) + return requested["review_id"] + + def test_full_workflow_gate_and_baseline(self) -> None: + self.issue() + self.submit_and_review() + linked = self.service.link_commit( + self.principals["lead"], + project_id="sandbox", + work_package_id="WP-001", + repository="git.example/project-bus", + commit_sha="0123456789abcdef", + relation="implementation", + idempotency_key="link-commit", + ) + self.assertEqual(linked["event"]["event_type"], "commit.linked") + gate = self.service.close_gate( + self.principals["pm"], + project_id="sandbox", + work_package_id="WP-001", + outcome="ACCEPT", + rationale="Criteria and review satisfied.", + idempotency_key="close-gate", + ) + self.assertEqual(gate["status"], "ACCEPTED") + baseline = self.service.baseline_release( + self.principals["pm"], + project_id="sandbox", + version="0.1.0", + artifact_uri="https://artifacts.example/project-bus-0.1.0.tar.gz", + digest="sha256:abc", + included_work_packages=["WP-001"], + idempotency_key="release-010", + ) + self.assertEqual(baseline["version"], "0.1.0") + with self.assertRaises(ConflictError): + self.service.close_gate( + self.principals["pm"], + project_id="sandbox", + work_package_id="WP-001", + outcome="HOLD", + rationale="Cannot reopen a terminal gate.", + idempotency_key="reopen-gate", + ) + with self.assertRaises(ConflictError): + self.service.baseline_release( + self.principals["pm"], + project_id="sandbox", + version="0.1.0", + artifact_uri="https://artifacts.example/other.tar.gz", + digest="sha256:different", + included_work_packages=["WP-001"], + idempotency_key="duplicate-release-version", + ) + events = self.service.sync_since( + self.principals["pa"], project_id="sandbox", cursor=0, limit=100 + ) + self.assertEqual(events["events"][-1]["event_type"], "baseline.released") + self.assertFalse(events["has_more"]) + + def test_pa_cannot_record_decision_or_close_gate(self) -> None: + with self.assertRaises(AuthorizationError): + self.service.record_decision( + self.principals["pa"], + project_id="sandbox", + decision_id="DEC-001", + kind="ARCHITECTURE", + title="Architecture", + decision="Use option A.", + rationale="Because.", + work_package_id=None, + idempotency_key="pa-decision", + ) + self.issue() + with self.assertRaises(AuthorizationError): + self.service.close_gate( + self.principals["pa"], + project_id="sandbox", + work_package_id="WP-001", + outcome="HOLD", + rationale="Waiting.", + idempotency_key="pa-gate", + ) + + def test_only_po_can_record_product_decision(self) -> None: + with self.assertRaises(AuthorizationError): + self.service.record_decision( + self.principals["pm"], + project_id="sandbox", + decision_id="DEC-PRODUCT-1", + kind="PRODUCT", + title="Product scope", + decision="Add a feature.", + rationale="Customer need.", + work_package_id=None, + idempotency_key="pm-product", + ) + result = self.service.record_decision( + self.principals["po"], + project_id="sandbox", + decision_id="DEC-PRODUCT-1", + kind="PRODUCT", + title="Product scope", + decision="Add a feature.", + rationale="Customer need.", + work_package_id=None, + idempotency_key="po-product", + ) + self.assertEqual(result["kind"], "PRODUCT") + + def test_reviewer_must_be_independent_and_have_review_role(self) -> None: + self.issue() + self.service.claim_work_package( + self.principals["lead"], + project_id="sandbox", + work_package_id="WP-001", + idempotency_key="claim", + ) + self.service.submit_result( + self.principals["lead"], + project_id="sandbox", + work_package_id="WP-001", + summary="Done.", + evidence=[], + idempotency_key="result", + ) + with self.assertRaises(ValidationError): + self.service.request_review( + self.principals["pa"], + project_id="sandbox", + work_package_id="WP-001", + reviewer_id="lead", + review_type="technical", + idempotency_key="self-review", + ) + + def test_gate_accept_requires_approved_review(self) -> None: + self.issue() + self.service.claim_work_package( + self.principals["lead"], + project_id="sandbox", + work_package_id="WP-001", + idempotency_key="claim", + ) + self.service.submit_result( + self.principals["lead"], + project_id="sandbox", + work_package_id="WP-001", + summary="Done.", + evidence=[], + idempotency_key="result", + ) + with self.assertRaises(ConflictError): + self.service.close_gate( + self.principals["pm"], + project_id="sandbox", + work_package_id="WP-001", + outcome="ACCEPT", + rationale="Premature.", + idempotency_key="premature-gate", + ) + + def test_idempotent_replay_and_key_mismatch(self) -> None: + first = self.service.issue_work_package( + self.principals["pm"], + project_id="sandbox", + work_package_id="WP-001", + title="Build it", + description="First request.", + acceptance_criteria=["Pass"], + idempotency_key="same-key", + ) + second = self.service.issue_work_package( + self.principals["pm"], + project_id="sandbox", + work_package_id="WP-001", + title="Build it", + description="First request.", + acceptance_criteria=["Pass"], + idempotency_key="same-key", + ) + self.assertFalse(first["idempotent_replay"]) + self.assertTrue(second["idempotent_replay"]) + self.assertEqual(first["event"]["event_id"], second["event"]["event_id"]) + with self.assertRaises(ConflictError): + self.service.issue_work_package( + self.principals["pm"], + project_id="sandbox", + work_package_id="WP-OTHER", + title="Different", + description="Different request.", + acceptance_criteria=["Pass"], + idempotency_key="same-key", + ) + + def test_concurrent_claim_has_single_winner(self) -> None: + self.issue() + barrier = threading.Barrier(2) + outcomes: list[str] = [] + lock = threading.Lock() + + def claim(actor_id: str) -> None: + barrier.wait() + try: + self.service.claim_work_package( + self.principals[actor_id], + project_id="sandbox", + work_package_id="WP-001", + idempotency_key=f"claim-{actor_id}", + ) + value = "won" + except ConflictError: + value = "lost" + with lock: + outcomes.append(value) + + first = threading.Thread(target=claim, args=("lead",)) + second = threading.Thread(target=claim, args=("impl2",)) + first.start() + second.start() + first.join() + second.join() + self.assertCountEqual(outcomes, ["won", "lost"]) + + def test_cursor_pagination_has_no_duplicates(self) -> None: + self.issue("WP-001") + self.issue("WP-002") + first = self.service.sync_since( + self.principals["pa"], project_id="sandbox", cursor=0, limit=3 + ) + second = self.service.sync_since( + self.principals["pa"], + project_id="sandbox", + cursor=first["next_cursor"], + limit=100, + ) + first_ids = {event["event_id"] for event in first["events"]} + second_ids = {event["event_id"] for event in second["events"]} + self.assertFalse(first_ids & second_ids) + self.assertTrue(first["has_more"]) + + def test_work_package_ids_are_project_scoped_and_reads_are_isolated(self) -> None: + self.issue("WP-001") + self.service.create_project( + self.system, + project_id="other", + name="Other", + idempotency_key="create-other", + ) + self.service.register_actor( + self.system, + project_id="other", + actor_id="pm", + auth_subject="subject-pm", + display_name="Pm", + role="PM", + idempotency_key="other-register-pm", + ) + self.service.issue_work_package( + self.principals["pm"], + project_id="other", + work_package_id="WP-001", + title="Same local identifier", + description="Different project.", + acceptance_criteria=["Isolated"], + idempotency_key="other-issue-wp", + ) + with self.assertRaises(AuthorizationError): + self.service.get_work_package( + self.principals["pa"], + project_id="other", + work_package_id="WP-001", + ) + + def test_event_log_rejects_update_and_delete(self) -> None: + self.issue() + with self.database.transaction(write=True) as connection: + event_id = connection.execute( + "SELECT event_id FROM event_log ORDER BY cursor LIMIT 1" + ).fetchone()["event_id"] + with self.assertRaises(sqlite3.IntegrityError): + connection.execute( + "UPDATE event_log SET event_type = 'tampered' WHERE event_id = ?", + (event_id,), + ) + with self.database.transaction(write=True) as connection: + with self.assertRaises(sqlite3.IntegrityError): + connection.execute("DELETE FROM event_log") + + def test_baseline_rejects_unaccepted_work_package(self) -> None: + self.issue() + with self.assertRaises(ConflictError): + self.service.baseline_release( + self.principals["pm"], + project_id="sandbox", + version="bad", + artifact_uri="artifact://bad", + digest="sha256:bad", + included_work_packages=["WP-001"], + idempotency_key="bad-baseline", + ) + + def test_escalation_can_be_raised_by_pa_but_only_resolved_by_pm(self) -> None: + escalation = self.service.raise_escalation( + self.principals["pa"], + project_id="sandbox", + severity="HIGH", + title="Decision needed", + description="Tolerance exceeded.", + target_role="PM", + idempotency_key="raise-escalation", + ) + with self.assertRaises(AuthorizationError): + self.service.resolve_escalation( + self.principals["pa"], + project_id="sandbox", + escalation_id=escalation["escalation_id"], + resolution="Resolved by PA.", + idempotency_key="pa-resolve", + ) + resolved = self.service.resolve_escalation( + self.principals["pm"], + project_id="sandbox", + escalation_id=escalation["escalation_id"], + resolution="PM supplied the required decision.", + idempotency_key="pm-resolve", + ) + self.assertEqual(resolved["status"], "RESOLVED") + + +if __name__ == "__main__": + unittest.main()