vepadmin/vep
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

P1 scaffold

Browse source
This commit is contained in:
ci 2026-08-07 19:26:49 +00:00
commit 372751ea49
14 changed files with 504 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.env.example Normal file
View file

@ -0,0 +1,5 @@
# Copy to .env and edit. Never commit .env.
PG_USER=vep
PG_PASS=change_me_to_a_real_secret
DATABASE_URL=postgresql://vep:change_me_to_a_real_secret@postgres:5432/vep
VEP_TENANT=t_default

39
.forgejo/workflows/ci.yml Normal file
View file

@ -0,0 +1,39 @@
# Forgejo Actions: P1 CI gate. Runs boundary tests + spine-check.
# Blocks merge if either fails (red). Safe: no infrastructure changes, just verification.
name: vep-p1-ci
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
verify:
runs-on: docker
container:
image: python:3.12-slim
services:
postgres:
image: pgvector/pgvector:0.7.0-pg16
env:
POSTGRES_USER: vep
POSTGRES_PASSWORD: vep_test
POSTGRES_DB: vep
ports:
- "5432:5432"
steps:
- uses: actions/checkout@v4
- name: Install uv
run: pip install uv && uv python install 3.12
- name: Install deps
run: uv pip install --system -r requirements.txt pytest httpx
- name: Init schema
env:
DATABASE_URL: postgresql://vep:vep_test@postgres:5432/vep
run: uv run python -m vep.initdb
- name: Boundary tests
env:
DATABASE_URL: postgresql://vep:vep_test@postgres:5432/vep
run: uv run pytest -q
- name: Spine-check
run: uv run python scripts/spine_check.py

10
.gitignore vendored Normal file
View file

@ -0,0 +1,10 @@
# VEP — git ignore (no secrets, no data, no artifacts)
.env
.venv/
__pycache__/
*.pyc
.pytest_cache/
.uv_cache/
*.log
data/
pgdata/

8
Dockerfile Normal file
View file

@ -0,0 +1,8 @@
FROM python:3.12-slim
ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
# Run schema init on container start (idempotent). Overridable for tests.
CMD ["sh", "-c", "python -m vep.initdb && uvicorn vep.main:app --host 0.0.0.0 --port 8000"]

48
compose.yml Normal file
View file

@ -0,0 +1,48 @@
# VEP P1 compose — pinned, extendable per phase. Internal network only.
# Postgres+pgvector not published to host. Caddy (host) proxies git.* and api.*.
services:
postgres:
image: pgvector/pgvector:0.7.0-pg16
environment:
POSTGRES_USER: ${PG_USER:-vep}
POSTGRES_PASSWORD: ${PG_PASS:-vep_dev_pass}
POSTGRES_DB: vep
volumes:
- pgdata:/var/lib/postgresql/data
- ./schema/init.sql:/docker-entrypoint-initdb.d/init.sql:ro
networks: [vep_net]
# NOT published to host — internal only.
app:
build:
context: .
dockerfile: Dockerfile
environment:
DATABASE_URL: postgresql://${PG_USER:-vep}:${PG_PASS:-vep_dev_pass}@postgres:5432/vep
VEP_TENANT: t_default
depends_on:
postgres:
condition: service_healthy
networks: [vep_net]
# NOT published — Caddy proxies localhost:8000.
forgejo:
image: codeberg.org/forgejo/forgejo:10.0.3
environment:
FORGEJO__database__DB_TYPE: sqlite
FORGEJO__server__SSH_PORT: 22
FORGEJO__server__ROOT_URL: https://git.mangoopsdesign.com
FORGEJO__server__APP_DATA_PATH: /data
ports:
- "2222:22" # host :2222 -> container ssh (avoids host :22)
volumes:
- forgejo_data:/data
networks: [vep_net]
networks:
vep_net:
driver: bridge
volumes:
pgdata:
forgejo_data:

3
pytest.ini Normal file
View file

@ -0,0 +1,3 @@
[pytest]
asyncio_mode = auto
testpaths = tests

7
requirements.txt Normal file
View file

@ -0,0 +1,7 @@
fastapi==0.115.6
uvicorn[standard]==0.34.0
asyncpg==0.30.0
pydantic==2.10.4
pytest==8.3.4
pytest-asyncio==0.24.0
httpx==0.28.1

92
schema/init.sql Normal file
View file

@ -0,0 +1,92 @@
-- VEP P1 schema: multi-tenant, append-only event_log, provenance, dead-letter,
-- actor model. Row-level security enforced from the first migration (JL-003 / #4).
-- Run idempotently by app startup against the postgres container.
-- Extensions
CREATE EXTENSION IF NOT EXISTS vector;
-- Tenant registry (commitment #4: tenancy at schema level day 1)
CREATE TABLE IF NOT EXISTS tenant (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Actor registry (agency first-class, JL-002): humans + digital agents, tier + audit
CREATE TABLE IF NOT EXISTS actor (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL REFERENCES tenant(id),
kind TEXT NOT NULL CHECK (kind IN ('human','digital')),
authority_tier INT NOT NULL CHECK (authority_tier BETWEEN 0 AND 3), -- T0..T3
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Append-only event log (commitment #2). No UPDATE/DELETE granted to app role.
-- create_capture: 'create' | 'capture' (new-market vs existing-demand probe)
-- confidence: 0..1 honest uncertainty
-- raw_ref: provenance pointer (source file / url / sheet cell / whatsapp msg id)
CREATE TABLE IF NOT EXISTS event_log (
id BIGSERIAL PRIMARY KEY,
tenant_id TEXT NOT NULL REFERENCES tenant(id),
actor_id TEXT NOT NULL, -- provenance (JL-002/#2)
schema_ver INT NOT NULL DEFAULT 1,
event_type TEXT NOT NULL,
payload JSONB NOT NULL,
create_capture TEXT CHECK (create_capture IN ('create','capture','unknown')),
confidence DOUBLE PRECISION CHECK (confidence BETWEEN 0 AND 1),
raw_ref TEXT, -- provenance (JL-002/#2)
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_event_log_tenant ON event_log(tenant_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_event_log_type ON event_log(tenant_id, event_type);
-- Dead-letter store for silent-loss prevention (JL-006/#2). Ingestion rejects land here.
CREATE TABLE IF NOT EXISTS dead_letter (
id BIGSERIAL PRIMARY KEY,
tenant_id TEXT,
reason TEXT NOT NULL,
raw_payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Row-level security: every table gated on tenant_id. App connects with a role that
-- can ONLY see its own tenant. DDL-level enforcement, not app-level.
ALTER TABLE event_log ENABLE ROW LEVEL SECURITY;
ALTER TABLE actor ENABLE ROW LEVEL SECURITY;
ALTER TABLE dead_letter ENABLE ROW LEVEL SECURITY;
-- The app connects as the role set below; we scope it per-tenant at connect time by
-- setting a session variable and matching it in the policy. (Simplified for P1 single
-- app connection; in P6 this moves to per-tenant JWT claims via Zitadel.)
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'vep_app') THEN
CREATE ROLE vep_app LOGIN PASSWORD 'vep_app'; -- overridden by .env in runtime
END IF;
END $$;
GRANT SELECT, INSERT ON ALL TABLES IN SCHEMA public TO vep_app;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO vep_app;
-- Explicitly DENY mutations on the log (append-only at the privilege layer)
REVOKE UPDATE, DELETE ON event_log FROM vep_app;
REVOKE UPDATE, DELETE ON dead_letter FROM vep_app;
-- RLS policy: app role only sees rows matching its session tenant.
DROP POLICY IF EXISTS tenant_isolation_event_log ON event_log;
CREATE POLICY tenant_isolation_event_log ON event_log
FOR ALL TO vep_app
USING (tenant_id = current_setting('app.current_tenant', true));
DROP POLICY IF EXISTS tenant_isolation_actor ON actor;
CREATE POLICY tenant_isolation_actor ON actor
FOR ALL TO vep_app
USING (tenant_id = current_setting('app.current_tenant', true));
DROP POLICY IF EXISTS tenant_isolation_dead_letter ON dead_letter;
CREATE POLICY tenant_isolation_dead_letter ON dead_letter
FOR ALL TO vep_app
USING (tenant_id IS NULL OR tenant_id = current_setting('app.current_tenant', true));
-- Seed a default tenant + a human T0 owner actor so the spine has a first row.
INSERT INTO tenant (id, name) VALUES ('t_default','Default Tenant')
ON CONFLICT (id) DO NOTHING;
INSERT INTO actor (id, tenant_id, kind, authority_tier)
VALUES ('a_owner','t_default','human',0)
ON CONFLICT (id) DO NOTHING;

80
scripts/spine_check.py Normal file
View file

@ -0,0 +1,80 @@
#!/usr/bin/env python3
"""VEP spine-check — merge gate (commitment #5).
For every file changed in the diff that looks like code (.py/.sql/.yml/.yaml/.js/.ts),
require evidence that it is justified by the spine:
1. references a JL id (e.g. JL-013 or #2) in the file (JL justification)
2. has a corresponding test entry OR is itself a test (value test)
3. names an owning agent/actor (from contracts: Owner, Builder, T0..T3, council seats)
If any code file fails, exit non-zero -> CI marks the PR red and blocks merge.
Usage: python scripts/spine_check.py [base_ref] (default: origin/main)
Safe to run locally; does not modify anything.
"""
from __future__ import annotations
import os
import re
import subprocess
import sys
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CODE_EXTS = (".py", ".sql", ".yml", ".yaml", ".js", ".ts")
JL_RE = re.compile(r"\b(JL-\d{3}|#[0-9]{1,2})\b")
AGENT_RE = re.compile(
r"\b(Owner|Builder|T0|T1|T2|T3|admin|actor|human|digital|tenant|analyst|advisor|finance|ops|commercial|people|data)\b",
re.IGNORECASE,
)
TEST_DIR_RE = re.compile(r"(^|/)tests?/", re.IGNORECASE)
TEST_FILE_RE = re.compile(r"(test_|_test\.|conftest)", re.IGNORECASE)
def changed_files(base: str) -> list[str]:
try:
out = subprocess.check_output(
["git", "diff", "--name-only", f"{base}...HEAD"], cwd=ROOT, text=True
)
except subprocess.CalledProcessError:
# fall back to working tree
out = subprocess.check_output(
["git", "diff", "--name-only"], cwd=ROOT, text=True
)
return [f.strip() for f in out.splitlines() if f.strip()]
def main() -> int:
base = sys.argv[1] if len(sys.argv) > 1 else "origin/main"
files = changed_files(base)
code_files = [f for f in files if f.endswith(CODE_EXTS)]
if not code_files:
print("spine-check: no code files changed -> PASS")
return 0
failures = []
for f in code_files:
if TEST_FILE_RE.search(f) or TEST_DIR_RE.search(f):
continue # tests are the value-proof themselves
path = os.path.join(ROOT, f)
if not os.path.exists(path):
continue
with open(path, encoding="utf-8", errors="replace") as fh:
text = fh.read()
problems = []
if not JL_RE.search(text):
problems.append("no JL justification (no JL-### or #n reference)")
if not AGENT_RE.search(text):
problems.append("no owning agent/actor named")
if problems:
failures.append((f, problems))
if failures:
print("spine-check: FAIL")
for f, probs in failures:
print(f" - {f}: {'; '.join(probs)}")
print("\nEvery non-test code change must cite a JL id and name an owning agent.")
return 1
print("spine-check: PASS (all changed code justified + agent-bound)")
return 0
if __name__ == "__main__":
sys.exit(main())

95
tests/test_boundary.py Normal file
View file

@ -0,0 +1,95 @@
"""VEP P1 boundary tests (canonical Part 6 discipline): test CONTRACTS, not internals.
These block merge (CI). Covers the four P1 acceptance criteria.
Run: cd /root/vep && uv run pytest -q
Requires DATABASE_URL to point at a fresh P1 postgres (the compose 'postgres' service).
"""
import os
import json
import pytest
import pytest_asyncio
import asyncpg
from httpx import AsyncClient, ASGITransport
from vep.main import app
from vep.initdb import init as db_init
from vep.db import connection
URL = os.environ.get("DATABASE_URL", "postgresql://vep:vep_dev_pass@postgres:5432/vep")
A_URL = URL.replace("vep", "vep_test_a")
B_URL = URL.replace("vep", "vep_test_b")
@pytest_asyncio.fixture(scope="module")
async def client():
await db_init()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
async def _seed_tenant(conn, tid, actor="a_owner"):
await conn.execute("INSERT INTO tenant (id,name) VALUES ($1,$2) ON CONFLICT DO NOTHING", tid, tid)
await conn.execute(
"INSERT INTO actor (id,tenant_id,kind,authority_tier) VALUES ($1,$2,'human',0) "
"ON CONFLICT DO NOTHING", actor, tid)
@pytest.mark.asyncio
async def test_tenant_isolation(client):
"""JL-003: tenant A cannot read tenant B's events."""
# Insert an event for tenant A under A's session scope.
async with connection("t_a") as conn:
await _seed_tenant(conn, "t_a", "a_a")
await conn.execute(
"INSERT INTO event_log (tenant_id, actor_id, event_type, payload, create_capture, confidence, raw_ref) "
"VALUES ($1,$2,$3,$4,$5,$6,$7)",
"t_a", "a_a", "test", json.dumps({"x": 1}), "create", 0.9, "src://a",
)
# Now B tries to read: RLS should hide A's row.
async with connection("t_b") as conn:
await _seed_tenant(conn, "t_b", "a_b")
n = await conn.fetchval("SELECT count(*) FROM event_log")
assert n == 0, "tenant B saw tenant A's events — RLS isolation failed"
@pytest.mark.asyncio
async def test_no_unvalidated_row(client):
"""JL-006: a missing actor_id is rejected AND recorded in dead_letter (no silent loss)."""
before = await _dead_count()
resp = await client.post("/events", json={"event_type": "no_actor", "payload": {}})
assert resp.status_code == 422, "invalid event was accepted"
after = await _dead_count()
assert after == before + 1, "rejected event was silently dropped"
@pytest.mark.asyncio
async def test_provenance_required(client):
"""#2: a valid event records actor_id + raw_ref + confidence in the log."""
async with connection("t_p") as conn:
await _seed_tenant(conn, "t_p", "a_p")
resp = await client.post("/events", json={
"tenant_id": "t_p", "actor_id": "a_p", "event_type": "probe",
"payload": {"v": 1}, "create_capture": "create", "confidence": 0.7, "raw_ref": "sheet://x",
})
assert resp.status_code == 200
async with connection("t_p") as conn:
row = await conn.fetchrow("SELECT actor_id, raw_ref, confidence FROM event_log WHERE id=$1", resp.json()["id"])
assert row["actor_id"] == "a_p" and row["raw_ref"] == "sheet://x" and row["confidence"] == 0.7
@pytest.mark.asyncio
async def test_append_only(client):
"""#2 / JL-006: the log cannot be mutated via the app role (privilege layer)."""
async with connection("t_ao") as conn:
await _seed_tenant(conn, "t_ao", "a_ao")
await conn.execute(
"INSERT INTO event_log (tenant_id, actor_id, event_type, payload) VALUES ($1,$2,$3,$4)",
"t_ao", "a_ao", "immutable", json.dumps({}),
)
with pytest.raises(Exception):
await conn.execute("UPDATE event_log SET event_type='hacked' WHERE tenant_id='t_ao'")
async def _dead_count():
async with connection() as conn:
return await conn.fetchval("SELECT count(*) FROM dead_letter")

0
vep/__init__.py Normal file
View file

25
vep/db.py Normal file
View file

@ -0,0 +1,25 @@
"""VEP P1 database access. App connects as the least-privilege 'vep_app' role so that
row-level security (RLS) policies actually apply (superuser/owner bypasses RLS).
A short-lived pool is created per connection context to avoid cross-event-loop reuse
bugs under pytest-asyncio. Volume is tiny in P1; this is fine.
"""
from __future__ import annotations
import os
import asyncpg
from contextlib import asynccontextmanager
DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://vep_app:vep_app@postgres:5432/vep")
VEP_TENANT = os.environ.get("VEP_TENANT", "t_default")
@asynccontextmanager
async def connection(tenant: str | None = None):
"""Yield a connection as vep_app with the tenant session var set (RLS enforcement)."""
conn = await asyncpg.connect(DATABASE_URL)
try:
await conn.execute(
"SELECT set_config('app.current_tenant', $1, false)", tenant or VEP_TENANT
)
yield conn
finally:
await conn.close()

25
vep/initdb.py Normal file
View file

@ -0,0 +1,25 @@
"""Run schema init idempotently. Compose mounts init.sql as a DDL init; this is the
explicit path used by the container CMD and by tests (in case initdb.d timing differs)."""
from __future__ import annotations
import os
import asyncpg
import pathlib
INIT_SQL = pathlib.Path(__file__).resolve().parent.parent / "schema" / "init.sql"
async def init() -> None:
# Run as the superuser (POSTGRES_USER) so roles/extensions/tables get created.
url = os.environ.get("DATABASE_ADMIN_URL", "postgresql://vep:vep_dev_pass@postgres:5432/vep")
conn = await asyncpg.connect(url)
try:
sql = INIT_SQL.read_text()
await conn.execute(sql)
print("VEP schema init OK")
finally:
await conn.close()
if __name__ == "__main__":
import asyncio
asyncio.run(init())

67
vep/main.py Normal file
View file

@ -0,0 +1,67 @@
"""VEP P1 API — health + event ingest (with contract enforcement + dead-letter).
JL anchors (spine-check will verify these on every change):
JL-002 Agency first-class: every event requires an actor_id.
JL-003 Multi-tenant day 1: events are tenant-scoped; RLS enforces isolation.
JL-006 No silent loss: an invalid event is written to dead_letter, never dropped.
#2 Provenance: event_log carries actor_id, raw_ref, confidence, create_capture.
"""
from __future__ import annotations
import os
from typing import Any, Literal
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field, ValidationError
from .db import connection, VEP_TENANT
app = FastAPI(title="VEP", version="0.1.0")
class EventIn(BaseModel):
tenant_id: str = Field(default=VEP_TENANT)
actor_id: str = Field(..., min_length=1) # JL-002 provenance
event_type: str = Field(..., min_length=1)
payload: dict[str, Any] = Field(default_factory=dict)
create_capture: Literal["create", "capture", "unknown"] | None = None
confidence: float | None = Field(default=None, ge=0.0, le=1.0) # honest uncertainty
raw_ref: str | None = None # #2 provenance
@app.get("/health")
async def health():
return {"status": "ok", "product": "VEP", "phase": "P1"}
@app.get("/events")
async def list_events(limit: int = 50):
async with connection() as conn:
rows = await conn.fetch(
"SELECT id, tenant_id, actor_id, event_type, create_capture, confidence, raw_ref, created_at "
"FROM event_log ORDER BY created_at DESC LIMIT $1",
limit,
)
return [dict(r) for r in rows]
@app.post("/events")
async def ingest(req: Request):
"""Contract: validate -> insert, or on failure write dead_letter (no silent loss)."""
raw = await req.json()
try:
ev = EventIn.model_validate(raw)
except ValidationError as e:
async with connection() as conn:
await conn.execute(
"INSERT INTO dead_letter (tenant_id, reason, raw_payload) VALUES ($1,$2,$3)",
raw.get("tenant_id"), f"validation: {e.errors()}", __import__("json").dumps(raw),
)
raise HTTPException(status_code=422, detail="rejected; recorded in dead_letter")
async with connection(tenant=ev.tenant_id) as conn:
row = await conn.fetchrow(
"INSERT INTO event_log (tenant_id, actor_id, event_type, payload, create_capture, confidence, raw_ref) "
"VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id, created_at",
ev.tenant_id, ev.actor_id, ev.event_type,
__import__("json").dumps(ev.payload), ev.create_capture, ev.confidence, ev.raw_ref,
)
return {"id": row["id"], "status": "recorded", "created_at": str(row["created_at"])}