diff --git a/README.md b/README.md index b5b9565..e35162d 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Prospect Intelligence Platform -A safety-first Phase 3 vertical slice for **manual**, evidence-led prospect qualification. It stores tenant-owned businesses and their child intelligence records, keeps provenance with each evidence item, supports a review pipeline, and records operationally relevant changes. **Automated discovery, DNS/website scanning, and outreach are not part of this release. Automated outreach is disabled.** +A safety-first Phase 4 design/implementation boundary for **manual**, evidence-led prospect qualification and the future job/live-log workflow. The current runtime remains the Phase 3 manual vertical slice: it stores tenant-owned businesses and child intelligence records, keeps provenance with each evidence item, supports a review pipeline, and records operationally relevant changes. **Automated discovery, DNS/website scanning, and outreach are not part of this release. Automated outreach is disabled.** ## Included @@ -11,8 +11,9 @@ A safety-first Phase 3 vertical slice for **manual**, evidence-led prospect qual - Responsive static dashboard under `apps/web` with authenticated explorer filters, paginated results, detail review, manual intake, notes/pipeline context, evidence provenance, and browser-only CSV preview. - Docker Compose runtime with non-root containers, read-only filesystems, health checks, and a named SQLite data volume. - Browser authentication with server-side sessions and an optional first-run admin bootstrap. +- Phase 4 MVP job monitor and SQLite-backed job/event schema/API surface, with the production limitations documented below. -## Phase 3 workflow +## Current workflow and Phase 4 boundary 1. A permitted workspace member manually creates or reviews a prospect. 2. The business detail response is the aggregate record for that tenant; related intelligence/evidence rows are returned only through the tenant-scoped detail surface. @@ -22,6 +23,14 @@ A safety-first Phase 3 vertical slice for **manual**, evidence-led prospect qual The API applies the organization/tenant boundary server-side to list, detail, child-record, notes, pipeline, and audit reads and writes. Clients must use the returned pagination metadata and follow `next`/`previous` links or tokens rather than assuming that one response contains the whole tenant dataset. See `apps/api/README.md` for the route contract and limits. +### Phase 4 jobs/live logging contract + +The planned asynchronous contract is: create one tenant-scoped job, return a stable job identifier, and move it through `queued` → `running` → a terminal state (`succeeded`, `failed`, `cancelled`). Each accepted request should carry an idempotency key whose scope and request fingerprint prevent duplicate jobs while allowing a safe replay of the original result. A job should persist append-only events with a monotonically increasing per-job sequence number, timestamp, level/type, safe message, and job/tenant identifiers. + +Clients should poll a tenant-scoped job status/events endpoint using `after_sequence` (or an equivalent cursor), with bounded backoff and terminal-state handling. SSE is a planned low-latency delivery option, not a current implementation; polling remains the compatibility fallback. Cancellation and retry must be explicit, authorized controls: cancellation is cooperative and may finish as `cancelled` or report that the job is already terminal; retry creates a new attempt while retaining the original job/idempotency lineage and must not duplicate side effects. + +The current MVP has SQLite job/event persistence, job status/list/detail and event APIs, cancellation/retry controls, and a browser job monitor that polls while work is active. SSE is not implemented; it remains a future delivery optimization over the persisted cursor. There is no Redis/Celery worker: the current in-process worker is suitable only for development/pilot use and must not be treated as durable, horizontally scalable execution. + ## Run locally ```bash @@ -69,7 +78,7 @@ Authenticated browser requests use a server-side session cookie; login creates a ## Explicit non-goals and remaining limitations -This Phase 3 release still has no automated discovery, DNS resolution, website/HTTP scanning, enrichment scheduler, external source adapter, email/SMS sender, or outreach endpoint. CSV remains a browser/API preview flow and does not silently persist rows. SQLite and the named local volume are suitable for the pilot only; there is no production migration runner, queue, or tested backup/restore command. The development password fallback is PBKDF2 rather than production Argon2id. Before production, complete the gates in `docs/SECURITY.md` and `docs/OPERATIONS.md`, including MFA, TLS, CSRF protection, rate limiting, durable audit retention, migrations, approved source policy, SSRF-safe fetching if a future scanner is approved, and tested backups/restores. +This Phase 4 boundary still has no automated discovery, DNS resolution, website/HTTP scanning, enrichment scheduler, external source adapter, email/SMS sender, or outreach endpoint. CSV remains a browser/API preview flow and does not silently persist rows. SQLite, the in-process worker, and the named local volume are suitable for the pilot only; production migration, durable queue/worker leases, event retention/backup, SSE delivery, and tested backup/restore remain unfinished. Redis and Celery are not implemented. The development password fallback is PBKDF2 rather than production Argon2id. Before production, complete the gates in `docs/SECURITY.md` and `docs/OPERATIONS.md`, including MFA, TLS, CSRF protection, rate limiting, tenant-scoped job/event authorization, idempotent side-effect handling, durable audit/event retention, migrations, approved source policy, SSRF-safe fetching if a future scanner is approved, and tested backups/restores. ## Verification diff --git a/apps/api/README.md b/apps/api/README.md index 1192e92..f2e895e 100644 --- a/apps/api/README.md +++ b/apps/api/README.md @@ -1,6 +1,6 @@ -# Prospect Platform API — Phase 3 +# Prospect Platform API — Phase 4 boundary -Dependency-light JSON API for tenant-scoped, **manual** prospect workflows. Core domain rules use Python's standard library and persistence is SQLite. The API stores businesses plus child intelligence/evidence records, pipeline state, notes, and audit context. It never performs automated discovery, DNS/website scanning, or outreach. +Dependency-light JSON API for tenant-scoped, **manual** prospect workflows. Core domain rules use Python's standard library and persistence is SQLite. The API stores businesses plus child intelligence/evidence records, pipeline state, notes, and audit context, with the Phase 4 job/live-log contract described below. It never performs automated discovery, DNS/website scanning, or outreach. ## Run @@ -18,6 +18,22 @@ Set `PROSPECT_API_PORT` or pass `--port`; set `PROSPECT_API_DB` or pass `--db` t All protected endpoints require the server-side session cookie. Every query is constrained by the authenticated user's `organization_id`; IDs from another tenant behave as not found and must not disclose whether a record exists. +## Phase 4 jobs and live logging (target contract) + +The intended job resource has a stable `job_id`, tenant/creator metadata, operation/payload fingerprint, `status`, `attempt`, timestamps, cancellation state, and terminal error/result metadata. Its lifecycle is `queued` → `running` → exactly one terminal state: `succeeded`, `failed`, or `cancelled`. State transitions and worker messages must be persisted transactionally with tenant and job identifiers; terminal jobs are immutable except for controlled retention/redaction. + +### Idempotency and events + +Mutating job creation should require an idempotency key (for example, an `Idempotency-Key` header). The key must be scoped to the authenticated tenant and operation, stored with a request fingerprint, and return the original job/result for an exact replay. Reuse with a different payload must be rejected rather than creating a second job. Idempotency must cover side effects, not merely the HTTP response. + +Live events should be append-only and ordered per job with a durable integer `sequence`/cursor. A consumer can resume from the last acknowledged sequence, tolerate duplicate delivery, and detect gaps. Events must contain only safe operational detail; do not persist passwords, session cookies, API keys, or unnecessary prospect/contact data. + +### Control and delivery semantics + +The MVP job routes are `POST /api/v1/jobs`, `GET /api/v1/jobs`, `GET /api/v1/jobs/{id}`, `GET /api/v1/jobs/{id}/events`, `POST /api/v1/jobs/{id}/cancel`, and `POST /api/v1/jobs/{id}/retry`. They are tenant-scoped and backed by SQLite persistence. Cancellation is best effort and must be race-safe with workers; retry creates a new attempt/lineage and must not repeat completed side effects. Polling supports a bounded cursor and backoff in the web monitor. An SSE endpoint may stream the same persisted sequence events with `Last-Event-ID`, heartbeats, disconnect/reconnect support, and polling fallback, but SSE is not implemented in this MVP. + +The current runtime has no durable job queue and no Redis/Celery integration. Its in-process/SQLite worker is single-instance, non-durable across process loss, and unsuitable for horizontal scaling or guaranteed execution; it is pilot-only. + ### Health and workspace - `GET /api/v1/health/live` — unauthenticated liveness check. @@ -58,6 +74,6 @@ All SQL uses parameters and all responses are JSON. Scores include `score_versio List and child-record endpoints are deliberately bounded. For business lists, use `page` (starting at 1) and `page_size` within the server-enforced maximum; invalid values are rejected rather than allowing an unbounded query. Supported filters are applied inside the tenant-scoped query before pagination: `q`, `score_min`, `score_max`, `website_class`, and `pipeline_stage`. The UI's page and filter controls are convenience clients, not authorization controls. A filtered page is not a count of the entire unfiltered tenant unless the response explicitly says so. -## Remaining limitations +## Remaining limitations and production migration work -SQLite is a pilot store with no production migration runner, queue, scheduler, durable backup command, or tested restore workflow. Authentication currently uses a development password fallback and does not by itself provide production Argon2id, MFA, CSRF protection, rate limiting, or a complete retention-grade audit system. Automated discovery, DNS/HTTP scanning, and outreach remain explicitly out of scope. +SQLite is a pilot store with an MVP job/event schema but no production migration runner, durable queue, scheduler, worker lease/recovery, durable backup command, or tested restore workflow. Authentication currently uses a development password fallback and does not by itself provide production Argon2id, MFA, CSRF protection, rate limiting, or a complete retention-grade audit system. Automated discovery, DNS/HTTP scanning, and outreach remain explicitly out of scope. Production migration work must add schema/index hardening and retention policy, tenant-scoped authorization tests, transactional event sequencing, cancellation/retry semantics, observability, and a reviewed Postgres plus durable queue/worker design. Redis/Celery remain future options, not implemented dependencies. diff --git a/apps/api/app/main.py b/apps/api/app/main.py index d7dbeca..e9868e6 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -1,5 +1,5 @@ from __future__ import annotations -import argparse, hashlib, json, os, re, secrets, sqlite3, sys +import argparse, hashlib, json, os, re, secrets, sqlite3, sys, threading, time from datetime import datetime, timedelta, timezone from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -15,8 +15,23 @@ SCHEMA = Path(__file__).resolve().parents[1] / "schema.sql" SESSION_DAYS = 7 PBKDF2_ITERATIONS = 300_000 MUTATING_ROLES = {"owner", "admin", "researcher"} +JOB_TYPES = {"noop", "prospect_recalculate"} +JOB_PAGE_SIZE = 100 +SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "credential", "private_key"} CHILD_TABLES = {"contacts": ("name", "email", "phone", "title", "do_not_contact"), "domains": ("domain", "kind"), "websites": ("url", "website_class"), "evidence": ("kind", "url", "claim"), "notes": ("body",)} +def redact(value): + if isinstance(value, dict): return {k: ("[REDACTED]" if str(k).lower() in SECRET_KEYS or any(s in str(k).lower() for s in ("password", "token", "secret", "api_key")) else redact(v)) for k,v in value.items()} + if isinstance(value, list): return [redact(v) for v in value[:100]] + if isinstance(value, str): return value[:2000] + return value + +def job_json(row): + result = row_json(row) + try: result["payload"] = json.loads(result.get("payload") or "{}") + except (ValueError, TypeError): result["payload"] = {} + return result + def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]: salt = salt or secrets.token_bytes(16); return hashlib.pbkdf2_hmac("sha256", password.encode(), salt, PBKDF2_ITERATIONS).hex(), salt.hex() def verify_password(password, encoded_hash, encoded_salt): @@ -95,6 +110,8 @@ class ApiHandler(BaseHTTPRequestHandler): if path=="/api/v1/dashboard/summary": row=db.execute("SELECT COUNT(*) businesses,COALESCE(AVG(score),0) average_score FROM businesses WHERE organization_id=?",(org,)).fetchone(); return self.send_json(200,{"organization_id":org,"businesses":row["businesses"],"average_score":round(row["average_score"],2),"suppressed":db.execute("SELECT COUNT(*) FROM suppressions WHERE organization_id=?",(org,)).fetchone()[0]}) if path=="/api/v1/businesses": return self.list_businesses(db,org,parse_qs(parsed.query)) + if path=="/api/v1/jobs": return self.list_jobs(db,org,parse_qs(parsed.query)) + if path.startswith("/api/v1/jobs/"): return self.get_job_route(db,org,path,parse_qs(parsed.query)) if path.startswith("/api/v1/businesses/"): bits=path.split("/"); ident=bits[4] if len(bits)>4 else "" if not ident.isdigit(): return self.send_json(404,{"error":"not_found"}) @@ -103,6 +120,59 @@ class ApiHandler(BaseHTTPRequestHandler): payload=row_json(row); payload.update(self.nested(db,int(ident),org)); return self.send_json(200,payload) return self.send_json(404,{"error":"not_found"}) finally: db.close() + def list_jobs(self, db, org, query): + try: limit=max(1,min(int(query.get("page_size",[50])[0]),JOB_PAGE_SIZE)); offset=max(0,int(query.get("offset",[0])[0])) + except (ValueError, TypeError): return self.send_json(400,{"error":"invalid_pagination"}) + rows=db.execute("SELECT * FROM jobs WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); more=len(rows)>limit + return self.send_json(200,{"organization_id":org,"items":[job_json(r) for r in rows[:limit]],"limit":limit,"offset":offset,"has_more":more}) + + def get_job_route(self, db, org, path, query): + bits=path.split("/") + if len(bits)<5 or not bits[4].isdigit(): return self.send_json(404,{"error":"not_found"}) + job=db.execute("SELECT * FROM jobs WHERE id=? AND organization_id=?",(int(bits[4]),org)).fetchone() + if not job:return self.send_json(404,{"error":"not_found"}) + if len(bits)==5:return self.send_json(200,job_json(job)) + if len(bits)==6 and bits[5]=="events": + try: after=max(0,int(query.get("after",[0])[0])) + except (ValueError,TypeError): return self.send_json(400,{"error":"invalid_sequence"}) + events=[row_json(r) for r in db.execute("SELECT * FROM job_events WHERE job_id=? AND organization_id=? AND sequence>? ORDER BY sequence",(job["id"],org,after))] + return self.send_json(200,{"items":events,"after":after}) + if len(bits)==7 and bits[5]=="events" and bits[6]=="stream": + try: after=max(0,int(query.get("after",[0])[0])) + except (ValueError,TypeError): return self.send_json(400,{"error":"invalid_sequence"}) + events=[row_json(r) for r in db.execute("SELECT * FROM job_events WHERE job_id=? AND organization_id=? AND sequence>? ORDER BY sequence",(job["id"],org,after))] + body=b"".join((b"event: "+str(e["event_type"]).encode()+b"\\ndata: "+json.dumps(e,sort_keys=True).encode()+b"\\n\\n") for e in events) + self.send_response(200);self.send_header("Content-Type","text/event-stream");self.send_header("Cache-Control","no-cache");self.send_header("Content-Length",str(len(body)));self.end_headers();self.wfile.write(body);return + return self.send_json(404,{"error":"not_found"}) + + def create_job(self, payload, db, user): + kind=str(payload.get("type","")).strip(); key=str(payload.get("idempotency_key","")).strip(); data=payload.get("payload",{}) + if kind not in JOB_TYPES:return self.send_json(400,{"error":"invalid_job_type"}) + if not key or len(key)>200 or not isinstance(data,dict):return self.send_json(400,{"error":"invalid_job_request"}) + safe=json.dumps(redact(data),sort_keys=True,separators=(",",":")); max_attempts=max(1,min(int(payload.get("max_attempts",3)),5)) if str(payload.get("max_attempts",3)).isdigit() else 3 + try: + cur=db.execute("INSERT INTO jobs(organization_id,idempotency_key,type,payload,max_attempts) VALUES(?,?,?,?,?)",(user["organization_id"],key,kind,safe,max_attempts)); jid=cur.lastrowid + self.add_job_event(db,jid,user["organization_id"],"queued","Job queued",0); self.audit(db,user,"job.created",str(jid)); db.commit(); getattr(self.server,"job_wakeup",threading.Event()).set() + return self.send_json(201,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone())) + except sqlite3.IntegrityError: + row=db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?",(user["organization_id"],key)).fetchone(); return self.send_json(200,job_json(row)) + + def add_job_event(self, db, jid, org, event_type, message, progress=0, error_code=None): + seq=db.execute("SELECT COALESCE(MAX(sequence),0)+1 FROM job_events WHERE job_id=?",(jid,)).fetchone()[0] + db.execute("INSERT INTO job_events(job_id,organization_id,sequence,event_type,message,progress,error_code) VALUES(?,?,?,?,?,?,?)",(jid,org,seq,event_type,str(message)[:500],progress,error_code)); return seq + + def job_action(self, db, user, path): + bits=path.split("/"); jid=int(bits[4]) if len(bits)>4 and bits[4].isdigit() else -1; action=bits[5] if len(bits)>5 else "" + job=db.execute("SELECT * FROM jobs WHERE id=? AND organization_id=?",(jid,user["organization_id"])).fetchone() + if not job:return self.send_json(404,{"error":"not_found"}) + if action=="cancel": + if job["status"] in ("queued","running"): db.execute("UPDATE jobs SET status='cancelled',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,));self.add_job_event(db,jid,user["organization_id"],"cancelled","Job cancelled",job["progress"]) + self.audit(db,user,"job.cancelled",str(jid));db.commit();return self.send_json(200,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone())) + if action=="retry": + if job["status"]!="failed":return self.send_json(409,{"error":"job_not_failed"}) + db.execute("UPDATE jobs SET status='queued',error_code=NULL,completed_at=NULL,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,));self.add_job_event(db,jid,user["organization_id"],"retry","Job retry queued",job["progress"]);self.audit(db,user,"job.retried",str(jid));db.commit();getattr(self.server,"job_wakeup",threading.Event()).set();return self.send_json(200,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone())) + return self.send_json(404,{"error":"not_found"}) + def list_businesses(self,db,org,query): def number(name, default=None): raw=query.get(name,[None])[0] @@ -132,8 +202,13 @@ class ApiHandler(BaseHTTPRequestHandler): c=SimpleCookie();c.load(self.headers.get("Cookie",""));t=c.get("session"); if t:db.execute("DELETE FROM sessions WHERE token_hash=?",(hashlib.sha256(t.value.encode()).hexdigest(),)) self.audit(db,user,"logout");db.commit();return self.send_json(200,{"ok":True},{"Set-Cookie":self.auth_cookie("",0)}) + if path=="/api/v1/jobs": + if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"}) + return self.create_job(self.read_json(),db,user) if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"}) payload=self.read_json(); org=user["organization_id"] + if path.startswith("/api/v1/jobs/"): + return self.job_action(db,user,path) if path=="/api/v1/businesses":return self.create_business(payload,db,user) if path=="/api/v1/suppressions":return self.create_suppression(payload,db,user) if path=="/api/v1/imports/preview":return self.preview_import(payload,db,org) @@ -213,8 +288,42 @@ class ApiHandler(BaseHTTPRequestHandler): return self.send_json(200,{"accepted":len(accepted),"duplicates":len(rows)-len(normalized)+len(normalized)-len(accepted)-suppressed,"suppressed":suppressed,"rows":accepted}) def log_message(self,*_):pass +def _job_worker(server): + while not server.job_stop.is_set(): + db=connect(server.db_path) + try: + job=db.execute("SELECT * FROM jobs WHERE status='queued' ORDER BY id LIMIT 1").fetchone() + if not job: + db.close(); server.job_wakeup.wait(.1); server.job_wakeup.clear(); continue + changed=db.execute("UPDATE jobs SET status='running',attempts=attempts+1,started_at=COALESCE(started_at,CURRENT_TIMESTAMP),updated_at=CURRENT_TIMESTAMP WHERE id=? AND status='queued'",(job["id"],)).rowcount + if not changed: db.close(); continue + db.commit(); org=job["organization_id"]; jid=job["id"]; server_handler=object.__new__(ApiHandler) + server_handler.add_job_event(db,jid,org,"started","Job started",0); db.commit() + try: payload=json.loads(job["payload"] or "{}") + except ValueError: payload={} + try: steps=1 if job["type"]=="noop" else max(1,min(int(payload.get("steps",5)),20)) + except (ValueError,TypeError): steps=5 + cancelled=False + for i in range(steps): + time.sleep(.01) + fresh=db.execute("SELECT status FROM jobs WHERE id=?",(jid,)).fetchone() + if not fresh or fresh["status"]=="cancelled": cancelled=True; break + progress=int((i+1)*100/steps); db.execute("UPDATE jobs SET progress=?,updated_at=CURRENT_TIMESTAMP WHERE id=? AND status='running'",(progress,jid)); server_handler.add_job_event(db,jid,org,"progress",f"Job progress {progress}%",progress); db.commit() + if cancelled: continue + if payload.get("force_fail") or (payload.get("fail_once") and job["attempts"] == 0): + db.execute("UPDATE jobs SET status='failed',error_code='DEMO_FAILURE',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,));server_handler.add_job_event(db,jid,org,"failed","Job failed",job["progress"],"DEMO_FAILURE") + else: + db.execute("UPDATE jobs SET status='succeeded',progress=100,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,));server_handler.add_job_event(db,jid,org,"succeeded","Job completed",100) + db.commit() + finally: db.close() + def create_server(host="127.0.0.1",port=8000,db_path="prospects.db"): - server=ThreadingHTTPServer((host,port),ApiHandler);server.db_path=db_path;connect(db_path).close();return server + server=ThreadingHTTPServer((host,port),ApiHandler);server.db_path=db_path;connect(db_path).close();server.job_stop=threading.Event();server.job_wakeup=threading.Event();server.job_thread=threading.Thread(target=_job_worker,args=(server,),daemon=True);server.job_thread.start() + original_close=server.server_close + def close(): + server.job_stop.set();server.job_wakeup.set();server.job_thread.join(timeout=2);original_close() + server.server_close=close + return server if __name__=="__main__": parser=argparse.ArgumentParser();parser.add_argument("--host",default="127.0.0.1");parser.add_argument("--port",type=int,default=int(os.environ.get("PROSPECT_API_PORT","8000")));parser.add_argument("--db",default=os.environ.get("PROSPECT_API_DB","prospects.db"));args=parser.parse_args();server=create_server(args.host,args.port,args.db);print(f"Prospect API listening on http://{args.host}:{args.port}",flush=True) try:server.serve_forever() diff --git a/apps/api/schema.sql b/apps/api/schema.sql index 5844a23..956004e 100644 --- a/apps/api/schema.sql +++ b/apps/api/schema.sql @@ -80,3 +80,37 @@ CREATE INDEX IF NOT EXISTS idx_pipeline_business ON pipeline_entries(business_id CREATE INDEX IF NOT EXISTS idx_pipeline_stage ON pipeline_entries(organization_id,stage); CREATE INDEX IF NOT EXISTS idx_notes_business ON notes(business_id,created_at); CREATE INDEX IF NOT EXISTS idx_interactions_business ON interactions(business_id,created_at); + +-- Phase 4 durable background jobs (additive-safe for existing databases). +CREATE TABLE IF NOT EXISTS jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + organization_id TEXT NOT NULL REFERENCES organizations(id), + idempotency_key TEXT NOT NULL, + type TEXT NOT NULL, + payload TEXT NOT NULL DEFAULT '{}', + status TEXT NOT NULL DEFAULT 'queued' CHECK(status IN ('queued','running','succeeded','failed','cancelled')), + attempts INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL DEFAULT 3, + progress INTEGER NOT NULL DEFAULT 0 CHECK(progress BETWEEN 0 AND 100), + error_code TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + started_at TEXT, + completed_at TEXT, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(organization_id,idempotency_key) +); +CREATE INDEX IF NOT EXISTS idx_jobs_org_created ON jobs(organization_id,created_at DESC,id DESC); +CREATE INDEX IF NOT EXISTS idx_jobs_queue ON jobs(status,created_at,id); +CREATE TABLE IF NOT EXISTS job_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + organization_id TEXT NOT NULL REFERENCES organizations(id), + sequence INTEGER NOT NULL, + event_type TEXT NOT NULL, + message TEXT NOT NULL DEFAULT '', + progress INTEGER NOT NULL DEFAULT 0 CHECK(progress BETWEEN 0 AND 100), + error_code TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(job_id,sequence) +); +CREATE INDEX IF NOT EXISTS idx_job_events_job_sequence ON job_events(job_id,sequence); diff --git a/apps/api/tests/test_jobs.py b/apps/api/tests/test_jobs.py new file mode 100644 index 0000000..89efcf6 --- /dev/null +++ b/apps/api/tests/test_jobs.py @@ -0,0 +1,82 @@ +import json +import os +import sqlite3 +import threading +import time +import unittest +from http.client import HTTPConnection +from tempfile import TemporaryDirectory + +from app.main import create_server, hash_password + + +class JobApiTests(unittest.TestCase): + def setUp(self): + self.tmp = TemporaryDirectory() + os.environ["BOOTSTRAP_ADMIN_EMAIL"] = "owner@example.test" + os.environ["BOOTSTRAP_ADMIN_PASSWORD"] = "development-password" + self.db_path = self.tmp.name + "/test.db" + self.server = create_server("127.0.0.1", 0, self.db_path) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start() + self.conn = HTTPConnection("127.0.0.1", self.server.server_port, timeout=3); self.cookie = None + self.request("POST", "/api/v1/auth/login", {"email":"owner@example.test", "password":"development-password"}) + + def tearDown(self): + self.server.shutdown(); self.server.server_close(); self.thread.join(timeout=2); self.tmp.cleanup() + + def request(self, method, path, payload=None): + body = json.dumps(payload).encode() if payload is not None else None + headers = {"Content-Type":"application/json"} if body else {} + if self.cookie: headers["Cookie"] = self.cookie + self.conn.request(method, path, body, headers); r = self.conn.getresponse() + c = r.getheader("Set-Cookie") + if c: self.cookie = c.split(";", 1)[0] + raw = r.read() + return r.status, (json.loads(raw) if raw and "json" in (r.getheader("Content-Type") or "") else raw.decode()) + + def test_create_is_idempotent_and_completes_with_sequence_events(self): + status, first = self.request("POST", "/api/v1/jobs", {"type":"noop", "payload":{"password":"hidden"}, "idempotency_key":"abc"}) + self.assertEqual(status, 201); self.assertEqual(first["status"], "queued") + self.assertNotIn("hidden", json.dumps(first)) + self.assertEqual(self.request("POST", "/api/v1/jobs", {"type":"noop", "payload":{"other":1}, "idempotency_key":"abc"})[1]["id"], first["id"]) + for _ in range(30): + status, job = self.request("GET", f"/api/v1/jobs/{first['id']}") + if job["status"] == "succeeded": break + time.sleep(.03) + self.assertEqual(job["status"], "succeeded") + status, events = self.request("GET", f"/api/v1/jobs/{first['id']}/events") + self.assertEqual(status, 200); seq = [e["sequence"] for e in events["items"]] + self.assertEqual(seq, list(range(1, len(seq)+1))); self.assertNotIn("hidden", json.dumps(events)) + + def test_cancel_retry_list_and_tenant_isolation(self): + _, job = self.request("POST", "/api/v1/jobs", {"type":"prospect_recalculate", "payload":{"steps":50}, "idempotency_key":"cancel-me"}) + status, cancelled = self.request("POST", f"/api/v1/jobs/{job['id']}/cancel"); self.assertEqual(status, 200) + self.assertIn(cancelled["status"], ("cancelled", "running")) + status, listing = self.request("GET", "/api/v1/jobs?page_size=2"); self.assertEqual(status, 200); self.assertLessEqual(len(listing["items"]), 2) + ph, salt = hash_password("other-password"); db = sqlite3.connect(self.db_path); db.execute("INSERT INTO organizations VALUES ('other-tenant','Other',CURRENT_TIMESTAMP)"); db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("other-tenant","other@example.test",ph,salt,"owner")); db.commit(); db.close() + self.cookie = None; self.request("POST", "/api/v1/auth/login", {"email":"other@example.test", "password":"other-password"}) + self.assertEqual(self.request("GET", f"/api/v1/jobs/{job['id']}")[0], 404) + + def test_failed_job_can_be_requeued_with_retry(self): + _, job = self.request("POST", "/api/v1/jobs", {"type":"noop", "payload":{"fail_once":True}, "idempotency_key":"retry-me"}) + for _ in range(30): + _, current = self.request("GET", f"/api/v1/jobs/{job['id']}") + if current["status"] == "failed": break + time.sleep(.03) + self.assertEqual(current["status"], "failed") + self.assertEqual(self.request("POST", f"/api/v1/jobs/{job['id']}/retry")[0], 200) + for _ in range(30): + _, current = self.request("GET", f"/api/v1/jobs/{job['id']}") + if current["status"] == "succeeded": break + time.sleep(.03) + self.assertEqual(current["status"], "succeeded") + + ph, salt = hash_password("viewer-password"); db = sqlite3.connect(self.db_path); db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("demo-tenant","viewer@example.test",ph,salt,"viewer")); db.commit(); db.close() + self.cookie = None; self.request("POST", "/api/v1/auth/login", {"email":"viewer@example.test", "password":"viewer-password"}) + self.assertEqual(self.request("POST", "/api/v1/jobs", {"type":"noop", "idempotency_key":"no"})[0], 403) + self.cookie = None; self.request("POST", "/api/v1/auth/login", {"email":"owner@example.test", "password":"development-password"}) + _, job = self.request("POST", "/api/v1/jobs", {"type":"noop", "idempotency_key":"audit"}) + db = sqlite3.connect(self.db_path); self.assertTrue(db.execute("SELECT 1 FROM audit_log WHERE action='job.created'").fetchone()); db.close() + + +if __name__ == "__main__": unittest.main() diff --git a/apps/web/README.md b/apps/web/README.md index af7f3ba..18d2265 100644 --- a/apps/web/README.md +++ b/apps/web/README.md @@ -1,6 +1,6 @@ -# ProspectOS web — Phase 3 +# ProspectOS web — Phase 4 boundary -Self-contained static frontend for the Prospect Platform API. There is no bundler or runtime dependency: serve this directory with any static HTTP server. The UI supports a manual, tenant-scoped review workflow; it does not discover prospects, scan DNS/websites, or send outreach. +Self-contained static frontend for the Prospect Platform API. There is no bundler or runtime dependency: serve this directory with any static HTTP server. The UI supports a manual, tenant-scoped review workflow and a Phase 4 MVP job monitor; it does not discover prospects, scan DNS/websites, or send outreach. ## Configure and run @@ -24,6 +24,14 @@ If not set, the UI uses `localStorage.prospect_api_base` when present; otherwise The API remains the source of truth for tenant isolation, pagination bounds, filters, pipeline transitions, notes, audit records, and suppression. See `apps/api/README.md` for the route contract. +## Phase 4 job/live-log UI contract + +A future job view should show `queued`, `running`, `succeeded`, `failed`, or `cancelled`, the current attempt, timestamps, safe error text, and a clear terminal state. It should display persisted events in sequence order, resume from the last cursor after refresh/reconnect, and tolerate duplicate events. Create/retry requests should send an idempotency key and show the returned job identity rather than starting duplicate work. + +The preferred live path is SSE backed by the persisted event cursor; polling with bounded backoff is the required fallback and should be used for browsers/proxies that do not support SSE. Cancel is a cooperative action with an explicit pending/terminal result; retry is available only when the API authorizes it and must be presented as a new attempt/lineage. The UI must never infer progress from timers or claim work completed because a request was accepted. + +The current client renders job status/counts, detail, structured errors, progress, and event timelines, and polls the jobs collection while queued/running work exists. It exposes authorized cancel/retry affordances based on the API response. There is no SSE client yet; polling is the current fallback and should remain available after SSE is introduced. The backend's SQLite/in-process worker is MVP-only. Do not add a Redis/Celery dependency by implication or label that worker production-ready. + ## Browser verification 1. Start the API from `apps/api` with `python3 app/main.py`. @@ -40,4 +48,4 @@ A zero-dependency static smoke page (`smoke-test.html`) checks the key DOM contr ## Remaining limitations -The static client has no background discovery, DNS/website scanner, enrichment scheduler, or outreach integration. It cannot make missing provenance authoritative and should display API-provided limitations rather than infer them. CSV preview is capped for display and is not an import workflow. Production deployment still requires the security and operations gates in `docs/SECURITY.md` and `docs/OPERATIONS.md`. +The static client has no background discovery, DNS/website scanner, enrichment scheduler, outreach integration, or SSE delivery. It cannot make missing provenance authoritative and should display API-provided limitations rather than infer them. CSV preview is capped for display and is not an import workflow. Production deployment still requires durable job execution, tenant-scoped controls, idempotency verification, SSE/polling verification, and the security and operations gates in `docs/SECURITY.md` and `docs/OPERATIONS.md`. diff --git a/apps/web/app.js b/apps/web/app.js index 110b145..4d517ac 100644 --- a/apps/web/app.js +++ b/apps/web/app.js @@ -15,7 +15,7 @@ async function request(path, options = {}) { const response=await fetch(endpoint(path),{...options,credentials:'include'}); if(response.status===401){showLogin('Your session has expired. Please sign in again.');throw new Error('unauthorized');} return response; } async function jsonRequest(path, options = {}) { const res=await request(path,options); const body=await res.json().catch(()=>({})); if(!res.ok)throw new Error(body.error||body.message||'Request failed'); return body; } function showLogin(message=''){currentUser=null;$('dashboardShell').hidden=true;$('loginScreen').hidden=false;$('loginMessage').textContent=message;$('loginMessage').className=`form-message${message?' error':''}`;} - function showDashboard(user){currentUser=user||{};const name=currentUser.name||currentUser.full_name||currentUser.email||'Workspace member';const role=currentUser.role||currentUser.roles?.[0]||'Member';$('userIdentity').textContent=`${name} · ${role}`;$('userAvatar').textContent=name.split(/\s+/).map(x=>x[0]).join('').slice(0,2).toUpperCase();$('loginScreen').hidden=true;$('dashboardShell').hidden=false;} + function showDashboard(user){currentUser=user||{};const name=currentUser.name||currentUser.full_name||currentUser.email||'Workspace member';const role=currentUser.role||currentUser.roles?.[0]||'Member';$('userIdentity').textContent=`${name} · ${role}`;$('userAvatar').textContent=name.split(/\s+/).map(x=>x[0]).join('').slice(0,2).toUpperCase();$('loginScreen').hidden=true;$('dashboardShell').hidden=false;updateJobPermissions();} function renderMetrics(summary){const total=Number(summary?.businesses??summary?.total??prospects.length),high=prospects.filter(p=>scoreFor(p)>=80).length,review=prospects.filter(p=>statusOf(p)==='review').length,fresh=prospects.length?Math.round(prospects.filter(p=>freshness(p).cls==='good').length/prospects.length*100):0;$('metricTotal').textContent=total;$('heroCount').textContent=`${total} prospects`;$('metricReview').textContent=summary?.needs_review??review;$('metricHigh').textContent=summary?.high_fit??high;$('metricFresh').textContent=`${summary?.freshness_under_7d??fresh}%`;} function filterValues(){return {q:$('searchInput').value.trim(),score:$('scoreFilter').value,status:$('statusFilter').value,website_class:$('websiteClassFilter').value,pipeline_stage:$('pipelineFilter').value};} function filtered(){const {q,score:sf,status:st,website_class:wc,pipeline_stage:ps}=filterValues();return prospects.filter(p=>{const s=scoreFor(p),text=`${p.name} ${p.website_domain||p.website||''} ${p.location||''}`.toLowerCase(),stage=p.pipeline_stage||p.pipeline?.stage||(Array.isArray(p.pipeline)?p.pipeline.at(-1)?.stage:'')||'new';return(!q||text.includes(q.toLowerCase()))&&(sf==='all'||(sf==='high'&&s>=80)||(sf==='medium'&&s>=60&&s<80)||(sf==='low'&&s<60))&&(st==='all'||statusOf(p)===st)&&(wc==='all'||(p.website_class||'missing')===wc)&&(ps==='all'||stage===ps);});} @@ -32,13 +32,44 @@ async function saveStage(form){const stage=new FormData(form).get('stage');if(!stage){message('pipelineMessage','Choose a pipeline stage.',true);return;}try{await jsonRequest(`/api/v1/businesses/${selectedId}/pipeline`,{method:'PATCH',headers:{'Content-Type':'application/json'},body:JSON.stringify({stage})});message('pipelineMessage','Pipeline stage updated.');await loadDetail(selectedId);await loadData();}catch(e){if(e.message!=='unauthorized')message('pipelineMessage',e.message,true);}} async function verify(){try{await jsonRequest(`/api/v1/businesses/${selectedId}/verify`,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({verified:true})});message('verifyMessage','Prospect marked verified.');await loadDetail(selectedId);await loadData();}catch(e){if(e.message!=='unauthorized')message('verifyMessage',e.message,true);}} async function addProspect(event){event.preventDefault();const data=Object.fromEntries(new FormData(event.currentTarget).entries());const msg=$('formMessage');try{const body=await jsonRequest('/api/v1/businesses',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(data)});prospects.unshift(body);msg.textContent='Added to review queue.';event.currentTarget.reset();renderMetrics(null);renderRows();}catch(e){if(e.message!=='unauthorized'){msg.textContent=e.message;msg.className='form-message error';}}} + let jobs = [], selectedJobId = null, jobPollTimer = null; + const jobStatuses = ['queued','running','succeeded','failed','cancelled']; + const jobRole = () => String(currentUser?.role || currentUser?.roles?.[0] || '').toLowerCase(); + const canManageJobs = () => ['admin','owner','operator','manager'].includes(jobRole()) || Boolean(currentUser?.permissions?.includes?.('jobs:manage')); + const jobStatus = job => String(job?.status || job?.state || 'queued').toLowerCase(); + const jobLabel = status => status.charAt(0).toUpperCase() + status.slice(1); + async function jobsRequest(path, options = {}) { return jsonRequest(path, options); } + function jobMessage(text, error = false) { const el = $('jobsMessage'); el.textContent = text || ''; el.className = `jobs-message${error ? ' error' : ''}`; } + function resetJobCounts() { jobStatuses.forEach(status => { const el = $(`jobCount${jobLabel(status)}`); if (el) el.textContent = '—'; }); } + function renderJobCounts(payload) { + const counts = payload?.counts || payload?.status_counts || payload?.summary || {}; + const derived = jobs.reduce((out, job) => { const status = jobStatus(job); if (jobStatuses.includes(status)) out[status] += 1; return out; }, {queued:0,running:0,succeeded:0,failed:0,cancelled:0}); + jobStatuses.forEach(status => { const value = counts[status] ?? counts[`${status}_count`] ?? derived[status]; $(`jobCount${jobLabel(status)}`).textContent = Number.isFinite(Number(value)) ? Number(value) : 0; }); + } + function renderJobsList() { + const list = $('jobsList'); + if (!jobs.length) { list.innerHTML = '
JOB DETAIL
ID ${esc(job.id)}
${esc(error.message || error.detail || error.description || JSON.stringify(error))}
${error.details ? `${esc(JSON.stringify(error.details, null, 2))}` : ''}No events returned yet.
'}${esc(error.message)}
No data rows found
';return;}const h=Object.keys(rows[0]);$('csvPreview').className='csv-table';$('csvPreview').innerHTML=`| ${esc(x)} | `).join('')}
|---|
| ${esc(r[x])} | `).join('')}
OPERATIONS
Track authenticated workspace jobs and their progress. No job data is shown until the API responds.
QUEUE
INTAKE
BULK INTAKE
Preview rows before adding them to your review queue.
No file selected
CSV stays in your browser until you confirm.