add crm pipeline and suppression center

apps/api/README.md

@@ -162,6 +162,20 @@ Clickable dashboard counts must link to the exact tenant-scoped predicate that p
 
 The current Phase 11 slice exposes `GET /api/v1/saved-filters`, `POST /api/v1/saved-filters`, `GET /api/v1/review-queue`, and `POST /api/v1/businesses/bulk-review`. Saved filters are durable and bounded, the queue is capped at 100 rows per request, and bulk verify/reject/assign accepts at most 100 explicit IDs. The slice remains pilot-grade: update/delete saved-filter handlers are not routed, review-queue results do not yet expose a full matching-set count or immutable filter snapshot, dashboard clickable-filter metadata is not a complete predicate, bulk execution has no preview/idempotency/per-record outcome contract, and bulk audit is one aggregate event. Do not infer stronger guarantees from the existing list filters.
 
+## Phase 12 CRM API contract
+
+Phase 12 introduces tenant-scoped CRM records for pipeline state, append-only interactions, normalized outcomes, bounded reports, and a suppression center. All routes must use the authenticated session's `organization_id`; a business, interaction, outcome, report, export, suppression, cursor, or filter ID from another tenant behaves as not found. The server—not the web client—enforces role permissions, state transitions, suppression, batch/report limits, and redaction.
+
+Pipeline transitions use `new` → `contacted` → `qualified` → `proposal` → `negotiation` → `won`/`lost`. Any paused/disqualified state must be explicitly configured before use and requires a reasoned, authorized reopen. The API accepts only policy-approved transitions, rejects direct jumps and changes to merged/inactive records, and appends actor, before/after state, reason, timestamp, and correlation/idempotency metadata. Same-state retries are idempotent. Reopening a terminal-for-now state creates a new event; it never edits history. Interactions do not implicitly advance the pipeline. `won` and `lost` require an explicit outcome/reason.
+
+Interaction records contain a bounded safe summary, channel, occurred/recorded timestamps, actor, business/contact reference, provenance, and idempotency lineage. Outcomes are normalized to `connected`, `no_answer`, `left_message`, `meeting_booked`, `meeting_held`, `qualified`, `disqualified`, `won`, `lost`, or `other`. `other` is an explicit catch-all, not proof of success or failure; missing data must not be silently assigned a negative outcome. Corrections append a superseding event and preserve the original. `do_not_contact` is safety-critical and cannot be overridden by score, stage, a later outcome, or a client payload.
+
+Reports must require bounded date ranges and page/row limits and expose their tenant, timezone, `as_of`, freshness, filter snapshot, and semantics. State reports use the latest effective state per active business; interaction/outcome reports use `occurred_at`; counts distinguish events from distinct businesses and page counts from matching-set counts. Suppressed, merged, inactive, and unknown records are labeled and never counted as contact-eligible. Report/export requests and results are audited, report caches are tenant-keyed, and report data must not become an authorization shortcut.
+
+Suppression endpoints accept only approved normalized identifier kinds and record source, reason, actor, scope, and timestamps. Matching occurs before writes, responses, exports, reports, caches, and queues. Suppressed contacts remain visible as `suppressed`/`do_not_contact` for safety review; deletion or unsuppression requires an authorized, reasoned, audited operation and does not retroactively rewrite interaction history. No endpoint sends messages, probes SMTP, performs validation mail, creates campaigns, or schedules delivery. Outreach remains disabled and requires a separate approved product/security/legal design.
+
+Every CRM mutation and report/export operation emits an audit record with tenant, actor, action, target, before/after or bounded result, policy/version, timestamps, correlation/idempotency ID, and safe reason. Audit and retention reads use the same organization predicate. CRM records, contact references, suppression decisions, report snapshots, and audit details require explicit retention classes, deletion/legal-hold semantics, and redacted logs. These are the Phase 12 contract; production readiness additionally requires durable migrations, worker/retry behavior, report reproducibility, export authorization, retention jobs, and transition/outcome/suppression/cross-tenant tests.
+
 ## Remaining limitations and production migration work
 
 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. Website scanning remains subject to pilot limits: no production egress proxy/isolation, distributed crawl coordinator, hardened resolver, or compliance-grade scan-history retention. Production work must add SSRF/DNS-rebinding/redirect-chain tests, egress policy, authenticated history/cache isolation, budget/abuse enforcement, durable result retention/deletion, observability, reviewed content/robots/terms policy, and the Phase 7 PSL/DNS/availability controls. Redis/Celery/Postgres remain future options, not implemented dependencies.

apps/api/app/main.py

@@ -58,6 +58,17 @@ def connect(db_path: str) -> sqlite3.Connection:
     for col, definition in (("verified", "INTEGER NOT NULL DEFAULT 0"), ("verified_at", "TEXT"), ("updated_at", "TEXT"), ("province", "TEXT NOT NULL DEFAULT ''"), ("city", "TEXT NOT NULL DEFAULT ''"), ("suburb", "TEXT NOT NULL DEFAULT ''"), ("merge_status", "TEXT NOT NULL DEFAULT 'active'"), ("merged_into_id", "INTEGER"), ("review_status", "TEXT NOT NULL DEFAULT 'pending'"), ("assigned_to", "TEXT NOT NULL DEFAULT ''"), ("review_metadata_json", "TEXT NOT NULL DEFAULT '{}'")):
         if col not in cols: db.execute(f"ALTER TABLE businesses ADD COLUMN {col} {definition}")
     db.execute("UPDATE businesses SET updated_at=COALESCE(updated_at,created_at) WHERE updated_at IS NULL")
+    # Phase 12 is additive-safe for databases created before CRM metadata existed.
+    for table, additions in {
+        "pipeline_entries": (("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT"), ("version", "INTEGER NOT NULL DEFAULT 1")),
+        "interactions": (("outcome", "TEXT NOT NULL DEFAULT 'other'"), ("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT")),
+        "suppressions": (("active", "INTEGER NOT NULL DEFAULT 1"), ("updated_at", "TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP"), ("actor_user_id", "INTEGER")),
+    }.items():
+        existing = {r[1] for r in db.execute(f"PRAGMA table_info({table})")}
+        for col, definition in additions:
+            if col not in existing: db.execute(f"ALTER TABLE {table} ADD COLUMN {col} {definition}")
+    db.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_pipeline_idempotency ON pipeline_entries(organization_id,idempotency_key) WHERE idempotency_key IS NOT NULL AND idempotency_key <> ''")
+    db.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_interaction_idempotency ON interactions(organization_id,idempotency_key) WHERE idempotency_key IS NOT NULL AND idempotency_key <> ''")
     db.execute("INSERT OR IGNORE INTO organizations (id,name) VALUES (?,?)", (ORGANIZATION_ID, "Demo organization"))
     for organization in db.execute("SELECT id FROM organizations").fetchall():
         for rule in DEFAULT_RULES:
@@ -106,8 +117,8 @@ class ApiHandler(BaseHTTPRequestHandler):
         db.execute("INSERT INTO audit_log (organization_id,user_id,action,details) VALUES (?,?,?,?)",(user["organization_id"],user["id"],action,details))
     def business(self, db, ident, org): return db.execute("SELECT * FROM businesses WHERE id=? AND organization_id=?",(ident,org)).fetchone()
     def nested(self, db, bid, org):
-        result={"contacts":[],"domains":[],"websites":[],"evidence":[],"pipeline":[],"notes":[]}
-        tables={"contacts":"contacts","domains":"domains","websites":"websites","evidence":"evidence","pipeline":"pipeline_entries","notes":"notes"}
+        result={"contacts":[],"domains":[],"websites":[],"evidence":[],"pipeline":[],"interactions":[],"notes":[]}
+        tables={"contacts":"contacts","domains":"domains","websites":"websites","evidence":"evidence","pipeline":"pipeline_entries","interactions":"interactions","notes":"notes"}
         for key, table in tables.items():
             result[key]=[row_json(r) for r in db.execute(f"SELECT * FROM {table} WHERE business_id=? AND organization_id=? ORDER BY id",(bid,org))]
         return result
@@ -181,7 +192,7 @@ class ApiHandler(BaseHTTPRequestHandler):
 
     def _review_item(self, row, db, org):
         item = row_json(row)
-        suppressed = is_suppressed(dict(row), [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?", (org,))])
+        suppressed = is_suppressed(dict(row), [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1", (org,))])
         merged = row["merge_status"] == "merged"
         item.update({"suppressed": bool(suppressed), "merged": bool(merged), "outreach_eligible": not suppressed and not merged and row["review_status"] not in ("rejected",), "review_flags": [x for x, yes in (("suppressed", suppressed), ("merged", merged), ("rejected", row["review_status"] == "rejected")) if yes]})
         return item
@@ -222,6 +233,69 @@ class ApiHandler(BaseHTTPRequestHandler):
         rows = db.execute("SELECT b.* FROM businesses b WHERE " + " AND ".join(where) + " ORDER BY b.score DESC,b.updated_at DESC,b.id DESC LIMIT ? OFFSET ?", params + [limit + 1, offset]).fetchall()
         return self.send_json(200, {"organization_id": org, "items": [self._review_item(r, db, org) for r in rows[:limit]], "limit": limit, "offset": offset, "has_more": len(rows) > limit})
 
+    OUTCOMES = {"connected", "no_answer", "left_message", "meeting_booked", "meeting_held", "qualified", "disqualified", "won", "lost", "other"}
+    STAGES = {"new", "contacted", "qualified", "proposal", "negotiation", "won", "lost"}
+
+    def _crm_suppressed(self, db, org, business):
+        return is_suppressed(dict(business), [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1", (org,))])
+
+    def _date_where(self, query, column, params):
+        start = (query.get("from") or query.get("start") or [""])[0]
+        end = (query.get("to") or query.get("end") or [""])[0]
+        for value in (start, end):
+            if value and (len(value) > 30 or not re.match(r"^\\d{4}-\\d{2}-\\d{2}(?:T[^ ]*)?$", value)):
+                raise ValueError("invalid_date")
+        if start and end:
+            try:
+                left=datetime.fromisoformat(start.replace("Z","+00:00")).date(); right=datetime.fromisoformat(end.replace("Z","+00:00")).date()
+                if (right-left).days > 366: raise ValueError("date_range_too_large")
+            except ValueError as exc:
+                if str(exc) == "date_range_too_large": raise
+                raise ValueError("invalid_date_range")
+        if start: params.append(start); clause = f"{column}>=?"
+        else: clause = ""
+        if end: params.append(end); clause += (" AND " if clause else "") + f"{column}<=?"
+        return clause
+
+    def list_pipeline(self, db, org, query):
+        params=[org]; where=["p.organization_id=?"]
+        bid=(query.get("business_id") or [""])[0]
+        if bid.isdigit(): where.append("p.business_id=?"); params.append(int(bid))
+        stage=(query.get("stage") or [""])[0]
+        if stage: where.append("p.stage=?"); params.append(stage)
+        rows=db.execute("SELECT p.* FROM pipeline_entries p WHERE " + " AND ".join(where) + " ORDER BY p.updated_at DESC,p.id DESC",params).fetchall()
+        return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows]})
+
+    def list_stages(self, db, org):
+        rows=db.execute("SELECT * FROM pipeline_stages WHERE organization_id=? AND active=1 ORDER BY position,id",(org,)).fetchall()
+        if not rows: return self.send_json(200,{"organization_id":org,"items":[{"name":x,"position":i,"active":True} for i,x in enumerate(("new","contacted","qualified","proposal","negotiation","won","lost"))]})
+        return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows]})
+
+    def list_interactions(self, db, org, query):
+        params=[org]; where=["i.organization_id=?"]
+        bid=(query.get("business_id") or [""])[0]
+        if bid.isdigit(): where.append("i.business_id=?"); params.append(int(bid))
+        outcome=(query.get("outcome") or [""])[0]
+        if outcome: where.append("i.outcome=?"); params.append(outcome)
+        rows=db.execute("SELECT i.* FROM interactions i WHERE " + " AND ".join(where) + " ORDER BY i.created_at DESC,i.id DESC",params).fetchall()
+        return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows]})
+
+    def list_suppressions(self, db, org, query):
+        rows=db.execute("SELECT * FROM suppressions WHERE organization_id=? ORDER BY id DESC",(org,)).fetchall()
+        return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows]})
+
+    def report(self, db, org, kind, query):
+        try:
+            if kind == "pipeline":
+                params=[org]; clause=self._date_where(query,"p.created_at",params); sql="SELECT p.stage, p.status, COUNT(*) count FROM pipeline_entries p WHERE p.organization_id=?" + ((" AND "+clause) if clause else "") + " GROUP BY p.stage,p.status ORDER BY p.stage,p.status"
+            elif kind == "outcomes":
+                params=[org]; clause=self._date_where(query,"i.created_at",params); sql="SELECT i.outcome, COUNT(*) count FROM interactions i WHERE i.organization_id=?" + ((" AND "+clause) if clause else "") + " GROUP BY i.outcome ORDER BY i.outcome"
+            else:
+                params=[org]; clause=self._date_where(query,"activity_at",params); sql="SELECT activity_type, COUNT(*) count FROM (SELECT 'pipeline' activity_type, created_at activity_at FROM pipeline_entries WHERE organization_id=? UNION ALL SELECT 'interaction',created_at FROM interactions WHERE organization_id=?) WHERE 1=1" + ((" AND "+clause) if clause else "") + " GROUP BY activity_type"
+                params=[org,org] + params[1:]
+            rows=db.execute(sql,params).fetchall(); return self.send_json(200,{"organization_id":org,"items":[dict(r) for r in rows]})
+        except ValueError as exc: return self.send_json(400,{"error":str(exc)})
+
     def do_GET(self):
         parsed=urlparse(self.path); path=parsed.path.rstrip("/")
         if path=="/api/v1/health/live": return self.send_json(200,{"status":"ok","organization_id":ORGANIZATION_ID})
@@ -252,6 +326,12 @@ class ApiHandler(BaseHTTPRequestHandler):
             if path=="/api/v1/domain-checks": return self.list_domain_checks(db,org,parse_qs(parsed.query))
             if path=="/api/v1/website-scans": return self.list_website_scans(db,org,parse_qs(parsed.query))
             if path=="/api/v1/contact-extractions": return self.list_contact_extractions(db,org,parse_qs(parsed.query))
+            if path=="/api/v1/pipeline-entries": return self.list_pipeline(db,org,parse_qs(parsed.query))
+            if path=="/api/v1/pipeline-stages": return self.list_stages(db,org)
+            if path=="/api/v1/outcomes": return self.send_json(200,{"items":sorted(self.OUTCOMES)})
+            if path=="/api/v1/interactions": return self.list_interactions(db,org,parse_qs(parsed.query))
+            if path=="/api/v1/suppressions": return self.list_suppressions(db,org,parse_qs(parsed.query))
+            if path in ("/api/v1/reports/pipeline","/api/v1/reports/outcomes","/api/v1/reports/activity"): return self.report(db,org,path.rsplit('/',1)[1],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 ""
@@ -371,7 +451,7 @@ class ApiHandler(BaseHTTPRequestHandler):
         key = str(payload.get("idempotency_key") or hashlib.sha256((str(scan_id or "") + source_url + source_html).encode()).hexdigest())[:200]
         existing = db.execute("SELECT * FROM contact_extractions WHERE organization_id=? AND business_id=? AND extraction_key=? ORDER BY id", (org, bid, key)).fetchall()
         if existing: return self.send_json(200, {"business_id": bid, "extraction_key": key, "items": [self._contact_json(r) for r in existing], "idempotent": True})
-        suppressions = [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?", (org,))]
+        suppressions = [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1", (org,))]
         try: found = extract_contacts(source_html, source_url, suppressions=suppressions, max_results=requested_limit)
         except ValueError as exc: return self.send_json(400, {"error": str(exc)})
         for item in found:
@@ -536,7 +616,7 @@ class ApiHandler(BaseHTTPRequestHandler):
         if drow:
             try: domain.update(json.loads(drow["result_json"] or "{}"))
             except (TypeError, ValueError): pass
-        suppressed = is_suppressed(dict(business), [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?", (org,))])
+        suppressed = is_suppressed(dict(business), [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1", (org,))])
         signals = signals_for_business(dict(business), website, contacts, domain, suppressed); self.ensure_score_rules(db, org); rules = [dict(r) for r in db.execute("SELECT * FROM score_rules WHERE organization_id=?", (org,))]; result = evaluate_score(signals, rules)
         override_score = payload.get("override_score"); override_eligible = payload.get("override_eligible")
         if override_score is not None or override_eligible is not None:
@@ -619,6 +699,9 @@ class ApiHandler(BaseHTTPRequestHandler):
             if path=="/api/v1/sources":return self.create_source(payload,db,user)
             if path=="/api/v1/discovery-queries":return self.create_query(payload,db,user)
             if path=="/api/v1/suppressions":return self.create_suppression(payload,db,user)
+            if path=="/api/v1/suppressions/import":return self.import_suppressions(payload,db,user)
+            if path=="/api/v1/pipeline-stages":return self.create_stage(payload,db,user)
+            if path=="/api/v1/interactions": return self.send_json(400,{"error":"business_id_required"})
             if path=="/api/v1/contact-extractions": return self.send_json(405,{"error":"method_not_allowed"})
             if path=="/api/v1/imports/preview":return self.preview_import(payload,db,org)
             if len(path.split("/"))==7 and path.split("/")[3:6]==["businesses",path.split("/")[4],"websites"] and path.split("/")[6]=="scan": return self.scan_business_website(int(path.split("/")[4]) if path.split("/")[4].isdigit() else -1,payload,db,user)
@@ -633,6 +716,8 @@ class ApiHandler(BaseHTTPRequestHandler):
                 return self.test_source(int(bits[4]),db,user) if bits[5]=="test" else self.ingest_source(int(bits[4]),payload,db,user)
             if len(bits)==6 and bits[3] == "discovery-queries" and bits[4].isdigit() and bits[5]=="run": return self.run_query(int(bits[4]),db,user)
             if len(bits)==7 and bits[:4]==["","api","v1","businesses"] and bits[5] in CHILD_TABLES and bits[6]=="": pass
+            if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5] == "pipeline": return self.update_pipeline(int(bits[4]) if bits[4].isdigit() else -1,payload,db,user)
+            if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5] == "interactions": return self.create_interaction(int(bits[4]) if bits[4].isdigit() else -1,payload,db,user)
             if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5] in CHILD_TABLES:return self.create_child(int(bits[4]) if bits[4].isdigit() else -1,bits[5],payload,db,user)
             if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5]=="verify":return self.verify_business(int(bits[4]) if bits[4].isdigit() else -1,payload,db,user)
             if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5]=="merge":return self.merge_business(int(bits[4]) if bits[4].isdigit() else -1,payload,db,user)
@@ -645,6 +730,10 @@ class ApiHandler(BaseHTTPRequestHandler):
             if not user:return
             if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"})
             bits=path.split("/")
+            if len(bits)==5 and bits[:4]==["","api","v1","pipeline-stages"] and bits[4].isdigit(): return self.update_stage(int(bits[4]),self.read_json(),db,user)
+            if len(bits)==5 and bits[:4]==["","api","v1","suppressions"] and bits[4].isdigit(): return self.update_suppression(int(bits[4]),self.read_json(),db,user)
+            if len(bits)==5 and bits[:4]==["","api","v1","pipeline-entries"] and bits[4].isdigit(): return self.update_pipeline_entry(int(bits[4]),self.read_json(),db,user)
+            if len(bits)==5 and bits[:4]==["","api","v1","interactions"] and bits[4].isdigit(): return self.update_interaction(int(bits[4]),self.read_json(),db,user)
             if len(bits)==5 and bits[:4]==["","api","v1","saved-filters"] and bits[4].isdigit(): return self.update_saved_filter(int(bits[4]),self.read_json(),db,user)
             if len(bits)==5 and bits[:4]==["","api","v1","score-rules"] and bits[4].isdigit(): return self.update_score_rule(int(bits[4]),self.read_json(),db,user)
             if len(bits)==5 and bits[:4]==["","api","v1","sources"] and bits[4].isdigit(): return self.update_source(int(bits[4]),self.read_json(),db,user)
@@ -658,6 +747,13 @@ class ApiHandler(BaseHTTPRequestHandler):
             if not user:return
             if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"})
             bits=path.split("/")
+            if len(bits)==5 and bits[:4]==["","api","v1","pipeline-stages"] and bits[4].isdigit(): return self.delete_crm_item("pipeline_stages",int(bits[4]),db,user)
+            if len(bits)==5 and bits[:4]==["","api","v1","pipeline-entries"] and bits[4].isdigit(): return self.delete_crm_item("pipeline_entries",int(bits[4]),db,user)
+            if len(bits)==5 and bits[:4]==["","api","v1","interactions"] and bits[4].isdigit(): return self.delete_crm_item("interactions",int(bits[4]),db,user)
+            if len(bits)==5 and bits[:4]==["","api","v1","suppressions"] and bits[4].isdigit():
+                row=db.execute("SELECT id FROM suppressions WHERE id=? AND organization_id=?",(int(bits[4]),user["organization_id"])).fetchone()
+                if not row:return self.send_json(404,{"error":"not_found"})
+                db.execute("DELETE FROM suppressions WHERE id=? AND organization_id=?",(int(bits[4]),user["organization_id"]));self.audit(db,user,"suppression.deleted",bits[4]);db.commit();return self.send_json(200,{"ok":True,"id":int(bits[4])})
             if len(bits)==5 and bits[:4]==["","api","v1","saved-filters"] and bits[4].isdigit(): return self.delete_saved_filter(int(bits[4]),db,user)
             return self.send_json(404,{"error":"not_found"})
         finally: db.close()
@@ -671,7 +767,7 @@ class ApiHandler(BaseHTTPRequestHandler):
     def create_business(self,payload,db,user):
         org=user["organization_id"]
         if not str(payload.get("name","")).strip():return self.send_json(400,{"error":"name_required"})
-        b=normalize_business(payload); suppressions=[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?",(org,))]
+        b=normalize_business(payload); suppressions=[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1",(org,))]
         if is_suppressed(b,suppressions):return self.send_json(409,{"error":"suppressed"})
         fields=[(c,b[c]) for c in ("website_domain","email","phone") if b[c]]
         if fields and db.execute("SELECT id FROM businesses WHERE organization_id=? AND ("+" OR ".join(f"{c}=?" for c,_ in fields)+")",[org]+[v for _,v in fields]).fetchone():return self.send_json(409,{"error":"duplicate"})
@@ -679,16 +775,16 @@ class ApiHandler(BaseHTTPRequestHandler):
     def create_suppression(self,payload,db,user):
         kind,value=payload.get("kind"),str(payload.get("value","")).strip().lower()
         if kind not in {"email","domain","phone"} or not value:return self.send_json(400,{"error":"invalid_suppression"})
-        try:db.execute("INSERT INTO suppressions(organization_id,kind,value) VALUES(?,?,?)",(user["organization_id"],kind,value))
+        try:db.execute("INSERT INTO suppressions(organization_id,kind,value,actor_user_id,active) VALUES(?,?,?,?,1)",(user["organization_id"],kind,value,user["id"]))
         except sqlite3.IntegrityError:pass
-        self.audit(db,user,"suppression.created",kind);db.commit();return self.send_json(201,dict(db.execute("SELECT * FROM suppressions WHERE organization_id=? AND kind=? AND value=?",(user["organization_id"],kind,value)).fetchone()))
+        self.audit(db,user,"suppression.created",kind);db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM suppressions WHERE organization_id=? AND kind=? AND value=?",(user["organization_id"],kind,value)).fetchone()))
     def child_business(self,db,bid,user):return self.business(db,bid,user["organization_id"])
     def create_child(self,bid,table,payload,db,user):
         if not self.child_business(db,bid,user):return self.send_json(404,{"error":"not_found"})
         if table=="contacts":
             email=str(payload.get("email","")).strip().lower(); phone=normalize_phone(payload.get("phone"));
             if email and not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$",email):return self.send_json(400,{"error":"invalid_contact"})
-            suppressed=is_suppressed({"email":email,"phone":phone},[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?",(user["organization_id"],))]); values=(str(payload.get("name","")).strip(),email,phone,str(payload.get("title","")).strip(),int(bool(payload.get("do_not_contact"))) or int(suppressed))
+            suppressed=is_suppressed({"email":email,"phone":phone},[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1",(user["organization_id"],))]); values=(str(payload.get("name","")).strip(),email,phone,str(payload.get("title","")).strip(),int(bool(payload.get("do_not_contact"))) or int(suppressed))
         elif table=="domains":
             value=normalize_domain(payload.get("domain"));
             if not value:return self.send_json(400,{"error":"invalid_domain"})
@@ -705,15 +801,100 @@ class ApiHandler(BaseHTTPRequestHandler):
             values=(str(payload["body"]).strip(),)
         columns=CHILD_TABLES[table]; db.execute(f"INSERT INTO {table}(business_id,organization_id,{','.join(columns)}) VALUES(?, ?, {','.join('?' for _ in columns)})",(bid,user["organization_id"])+values); rid=db.execute("SELECT last_insert_rowid()").fetchone()[0];self.audit(db,user,f"{table}.created",str(rid));db.commit();return self.send_json(201,row_json(db.execute(f"SELECT * FROM {table} WHERE id=?",(rid,)).fetchone()))
     def update_pipeline(self,bid,payload,db,user):
-        if not self.child_business(db,bid,user) or not str(payload.get("stage","")).strip():return self.send_json(404 if not self.child_business(db,bid,user) else 400,{"error":"not_found" if not self.child_business(db,bid,user) else "stage_required"})
-        stage=str(payload["stage"]).strip();status=str(payload.get("status","active")).strip() or "active";db.execute("INSERT INTO pipeline_entries(business_id,organization_id,stage,status) VALUES(?,?,?,?)",(bid,user["organization_id"],stage,status));rid=db.execute("SELECT last_insert_rowid()").fetchone()[0];self.audit(db,user,"pipeline.updated",stage);db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM pipeline_entries WHERE id=?",(rid,)).fetchone()))
+        if not self.child_business(db,bid,user): return self.send_json(404,{"error":"not_found"})
+        stage=str(payload.get("stage","")).strip(); status=str(payload.get("status","active")).strip() or "active"
+        if not stage or stage not in self.STAGES or status not in {"active","won","lost","paused"}: return self.send_json(400,{"error":"invalid_pipeline"})
+        key=str(payload.get("idempotency_key","")).strip(); org=user["organization_id"]
+        if key:
+            prior=db.execute("SELECT * FROM pipeline_entries WHERE organization_id=? AND idempotency_key=?",(org,key)).fetchone()
+            if prior:return self.send_json(200,row_json(prior))
+        suppressed=self._crm_suppressed(db,org,self.business(db,bid,org))
+        if suppressed:return self.send_json(409,{"error":"do_not_contact","outreach_disabled":True})
+        cur=db.execute("INSERT INTO pipeline_entries(business_id,organization_id,stage,status,notes,next_action,follow_up_at,actor_user_id,idempotency_key) VALUES(?,?,?,?,?,?,?,?,?)",(bid,org,stage,status,str(payload.get("notes",payload.get("body","")))[:5000],str(payload.get("next_action",""))[:500],payload.get("follow_up_at"),user["id"],key or None)); rid=cur.lastrowid
+        self.audit(db,user,"pipeline.created",str(rid)); db.commit(); return self.send_json(200 if getattr(self,"command","")=="PATCH" else 201,row_json(db.execute("SELECT * FROM pipeline_entries WHERE id=?",(rid,)).fetchone()))
+
+    def update_pipeline_entry(self,eid,payload,db,user):
+        org=user["organization_id"]; row=db.execute("SELECT * FROM pipeline_entries WHERE id=? AND organization_id=?",(eid,org)).fetchone()
+        if not row:return self.send_json(404,{"error":"not_found"})
+        if payload.get("expected_updated_at") and payload["expected_updated_at"] != row["updated_at"]: return self.send_json(409,{"error":"conflict"})
+        values={k:payload[k] for k in ("stage","status","notes","next_action","follow_up_at") if k in payload}
+        if "stage" in values and values["stage"] not in self.STAGES:return self.send_json(400,{"error":"invalid_stage"})
+        if "status" in values and values["status"] not in {"active","won","lost","paused"}:return self.send_json(400,{"error":"invalid_status"})
+        if not values:return self.send_json(400,{"error":"no_changes"})
+        cols=[]; args=[]
+        for k,v in values.items():cols.append(k+"=?");args.append(v)
+        args += [eid,org]; db.execute("UPDATE pipeline_entries SET "+",".join(cols)+",version=version+1,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",args); self.audit(db,user,"pipeline.updated",str(eid));db.commit()
+        return self.send_json(200,row_json(db.execute("SELECT * FROM pipeline_entries WHERE id=?",(eid,)).fetchone()))
+
+    def create_interaction(self,bid,payload,db,user):
+        org=user["organization_id"]; business=self.business(db,bid,org)
+        if not business:return self.send_json(404,{"error":"not_found"})
+        outcome=str(payload.get("outcome","other")).strip().lower(); kind=str(payload.get("kind","")).strip().lower()
+        if not kind or outcome not in self.OUTCOMES:return self.send_json(400,{"error":"invalid_interaction"})
+        if self._crm_suppressed(db,org,business):return self.send_json(409,{"error":"do_not_contact","outreach_disabled":True})
+        key=str(payload.get("idempotency_key","")).strip()
+        if key:
+            prior=db.execute("SELECT * FROM interactions WHERE organization_id=? AND idempotency_key=?",(org,key)).fetchone()
+            if prior:return self.send_json(200,row_json(prior))
+        cur=db.execute("INSERT INTO interactions(business_id,organization_id,kind,body,outcome,notes,next_action,follow_up_at,actor_user_id,idempotency_key) VALUES(?,?,?,?,?,?,?,?,?,?)",(bid,org,kind,str(payload.get("body",payload.get("notes","")))[:5000],outcome,str(payload.get("notes",""))[:5000],str(payload.get("next_action",""))[:500],payload.get("follow_up_at"),user["id"],key or None)); self.audit(db,user,"interaction.created",str(cur.lastrowid));db.commit()
+        return self.send_json(201,row_json(db.execute("SELECT * FROM interactions WHERE id=?",(cur.lastrowid,)).fetchone()))
+
+    def update_interaction(self,iid,payload,db,user):
+        org=user["organization_id"]; row=db.execute("SELECT * FROM interactions WHERE id=? AND organization_id=?",(iid,org)).fetchone()
+        if not row:return self.send_json(404,{"error":"not_found"})
+        values={k:payload[k] for k in ("kind","body","outcome","notes","next_action","follow_up_at") if k in payload}
+        if "outcome" in values and values["outcome"] not in self.OUTCOMES:return self.send_json(400,{"error":"invalid_outcome"})
+        if not values:return self.send_json(400,{"error":"no_changes"})
+        cols=[];args=[]
+        for k,v in values.items():cols.append(k+"=?");args.append(v)
+        args += [iid,org];db.execute("UPDATE interactions SET "+",".join(cols)+",updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",args);self.audit(db,user,"interaction.updated",str(iid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM interactions WHERE id=?",(iid,)).fetchone()))
+
+    def update_suppression(self,sid,payload,db,user):
+        row=db.execute("SELECT * FROM suppressions WHERE id=? AND organization_id=?",(sid,user["organization_id"])).fetchone()
+        if not row:return self.send_json(404,{"error":"not_found"})
+        if "active" not in payload:return self.send_json(400,{"error":"active_required"})
+        db.execute("UPDATE suppressions SET active=?,actor_user_id=?,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(int(bool(payload["active"])),user["id"],sid,user["organization_id"]));self.audit(db,user,"suppression.updated",str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM suppressions WHERE id=?",(sid,)).fetchone()))
+
+    def import_suppressions(self,payload,db,user):
+        items=payload.get("items",payload.get("suppressions"));
+        if not isinstance(items,list) or len(items)>1000:return self.send_json(400,{"error":"invalid_suppression_import"})
+        imported=0
+        for item in items:
+            if not isinstance(item,dict) or item.get("kind") not in {"email","domain","phone"} or not str(item.get("value","")).strip():return self.send_json(400,{"error":"invalid_suppression"})
+            kind=item["kind"];value=str(item["value"]).strip().lower(); cur=db.execute("INSERT OR IGNORE INTO suppressions(organization_id,kind,value,actor_user_id) VALUES(?,?,?,?)",(user["organization_id"],kind,value,user["id"]));imported += cur.rowcount
+        self.audit(db,user,"suppressions.imported",str(imported));db.commit();return self.send_json(201,{"imported":imported,"received":len(items)})
+
+    def create_stage(self,payload,db,user):
+        name=str(payload.get("name","")).strip().lower()
+        if not name or len(name)>80 or not re.match(r"^[a-z0-9_-]+$",name):return self.send_json(400,{"error":"invalid_stage"})
+        try: position=int(payload.get("position",0))
+        except (TypeError,ValueError):return self.send_json(400,{"error":"invalid_stage"})
+        try:cur=db.execute("INSERT INTO pipeline_stages(organization_id,name,position) VALUES(?,?,?)",(user["organization_id"],name,position))
+        except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_stage"})
+        self.audit(db,user,"pipeline_stage.created",str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM pipeline_stages WHERE id=?",(cur.lastrowid,)).fetchone()))
+
+    def update_stage(self,sid,payload,db,user):
+        row=db.execute("SELECT * FROM pipeline_stages WHERE id=? AND organization_id=?",(sid,user["organization_id"])).fetchone()
+        if not row:return self.send_json(404,{"error":"not_found"})
+        values={k:payload[k] for k in ("name","position","active") if k in payload}
+        if not values:return self.send_json(400,{"error":"no_changes"})
+        if "name" in values and (not isinstance(values["name"],str) or not re.match(r"^[a-z0-9_-]+$",values["name"])):return self.send_json(400,{"error":"invalid_stage"})
+        cols=[];args=[]
+        for k,v in values.items():cols.append(k+"=?");args.append(int(bool(v)) if k=="active" else v)
+        args += [sid,user["organization_id"]];db.execute("UPDATE pipeline_stages SET "+",".join(cols)+",updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",args);self.audit(db,user,"pipeline_stage.updated",str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM pipeline_stages WHERE id=?",(sid,)).fetchone()))
+
+    def delete_crm_item(self,table,ident,db,user):
+        row=db.execute(f"SELECT id FROM {table} WHERE id=? AND organization_id=?",(ident,user["organization_id"])).fetchone()
+        if not row:return self.send_json(404,{"error":"not_found"})
+        db.execute(f"DELETE FROM {table} WHERE id=? AND organization_id=?",(ident,user["organization_id"]));self.audit(db,user,table+".deleted",str(ident));db.commit();return self.send_json(200,{"ok":True,"id":ident})
+
     def verify_business(self,bid,payload,db,user):
         if not self.child_business(db,bid,user):return self.send_json(404,{"error":"not_found"})
         verified=bool(payload.get("verified",True)); now=datetime.now(timezone.utc).replace(microsecond=0).isoformat();db.execute("UPDATE businesses SET verified=?,verified_at=?,review_status=?,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(int(verified),now if verified else None,"verified" if verified else "pending",bid,user["organization_id"]));self.audit(db,user,"business.verified",str(verified));db.commit();row=self.business(db,bid,user["organization_id"]);return self.send_json(200,row_json(row))
     def preview_import(self,payload,db,org):
         rows=payload.get("rows",[])
         if not isinstance(rows,list):return self.send_json(400,{"error":"rows_required"})
-        normalized=deduplicate_businesses([r for r in rows if isinstance(r,dict) and str(r.get("name","")).strip()]); suppressions=[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?",(org,))];existing=[row_json(r) for r in db.execute("SELECT * FROM businesses WHERE organization_id=?",(org,))];seen=set();accepted=[];suppressed=0;existing_keys={deduplication_key(x) for x in existing}
+        normalized=deduplicate_businesses([r for r in rows if isinstance(r,dict) and str(r.get("name","")).strip()]); suppressions=[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1",(org,))];existing=[row_json(r) for r in db.execute("SELECT * FROM businesses WHERE organization_id=?",(org,))];seen=set();accepted=[];suppressed=0;existing_keys={deduplication_key(x) for x in existing}
         for b in normalized:
             key=deduplication_key(b)
             if is_suppressed(b,suppressions):suppressed+=1

apps/api/schema.sql

@@ -89,6 +89,16 @@ CREATE INDEX IF NOT EXISTS idx_pipeline_stage ON pipeline_entries(organization_i
 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 12 CRM workflow metadata and safe, auditable state.
+CREATE TABLE IF NOT EXISTS pipeline_stages (
+ id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+ name TEXT NOT NULL, position INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1,
+ created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ UNIQUE(organization_id,name)
+);
+CREATE INDEX IF NOT EXISTS idx_pipeline_stages_org ON pipeline_stages(organization_id,position,id);
+CREATE INDEX IF NOT EXISTS idx_interactions_org_created ON interactions(organization_id,created_at DESC,id DESC);
+
 -- Phase 4 durable background jobs (additive-safe for existing databases).
 CREATE TABLE IF NOT EXISTS jobs (
  id INTEGER PRIMARY KEY AUTOINCREMENT,

apps/api/tests/test_phase12_api.py

@@ -0,0 +1,42 @@
+import json
+import os
+import threading
+import unittest
+from http.client import HTTPConnection
+from tempfile import TemporaryDirectory
+
+from app.main import create_server
+
+
+class Phase12Tests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory(); os.environ['BOOTSTRAP_ADMIN_EMAIL']='p12@example.test'; os.environ['BOOTSTRAP_ADMIN_PASSWORD']='password-p12'
+        self.server=create_server('127.0.0.1',0,self.tmp.name+'/db.sqlite'); 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); self.cookie=None
+        self.request('POST','/api/v1/auth/login',{'email':'p12@example.test','password':'password-p12'})
+    def tearDown(self):
+        self.server.shutdown(); self.server.server_close(); self.thread.join(timeout=2); os.environ.pop('BOOTSTRAP_ADMIN_EMAIL',None); os.environ.pop('BOOTSTRAP_ADMIN_PASSWORD',None); self.tmp.cleanup()
+    def request(self,m,p,b=None):
+        h={'Content-Type':'application/json'} if b is not None else {}; h['Cookie']=self.cookie if self.cookie else ''
+        self.conn.request(m,p,json.dumps(b).encode() if b is not None else None,h); r=self.conn.getresponse(); sc=r.getheader('Set-Cookie');
+        if sc: self.cookie=sc.split(';',1)[0]
+        return r.status,json.loads(r.read() or b'{}')
+    def test_pipeline_interaction_transition_and_no_contact(self):
+        s,b=self.request('POST','/api/v1/businesses',{'name':'CRM Co','email':'crm@example.test'}); self.assertEqual(s,201)
+        s,p=self.request('POST',f"/api/v1/businesses/{b['id']}/pipeline",{'stage':'new','status':'active','notes':'first','next_action':'qualify','follow_up_at':'2026-09-10T10:00:00Z','idempotency_key':'p1'}); self.assertEqual(s,201); self.assertEqual(p['actor_user_id'],1)
+        s,p2=self.request('PATCH',f"/api/v1/pipeline-entries/{p['id']}",{'stage':'qualified','expected_updated_at':p['updated_at']}); self.assertEqual(s,200); self.assertEqual(p2['stage'],'qualified')
+        s,i=self.request('POST',f"/api/v1/businesses/{b['id']}/interactions",{'kind':'call','outcome':'connected','notes':'good fit','next_action':'demo'}); self.assertEqual(s,201); self.assertEqual(i['outcome'],'connected')
+        self.assertEqual(self.request('GET','/api/v1/reports/pipeline')[0],200); self.assertEqual(self.request('GET','/api/v1/reports/outcomes')[0],200); self.assertEqual(self.request('GET','/api/v1/reports/activity')[0],200)
+    def test_suppression_center_bulk_and_enforces_pipeline_interaction(self):
+        s,b=self.request('POST','/api/v1/businesses',{'name':'Blocked CRM','email':'blocked@example.test'}); self.assertEqual(s,201)
+        s,x=self.request('POST','/api/v1/suppressions/import',{'items':[{'kind':'email','value':'blocked@example.test'},{'kind':'domain','value':'bad.example'}]}); self.assertEqual(s,201); self.assertEqual(x['imported'],2)
+        s,l=self.request('GET','/api/v1/suppressions'); self.assertEqual(s,200); self.assertEqual(len(l['items']),2)
+        self.assertEqual(self.request('POST',f"/api/v1/businesses/{b['id']}/pipeline",{'stage':'new'})[0],409)
+        self.assertEqual(self.request('POST',f"/api/v1/businesses/{b['id']}/interactions",{'kind':'call','outcome':'connected'})[0],409)
+        sid=l['items'][0]['id']; self.assertEqual(self.request('PATCH',f'/api/v1/suppressions/{sid}',{'active':False})[0],200)
+        self.assertEqual(self.request('DELETE',f'/api/v1/suppressions/{sid}')[0],200)
+    def test_validation_and_date_bounds(self):
+        s,b=self.request('POST','/api/v1/businesses',{'name':'Validation CRM'}); self.assertEqual(s,201)
+        self.assertEqual(self.request('POST',f"/api/v1/businesses/{b['id']}/interactions",{'kind':'call','outcome':'not-a-real-outcome'})[0],400)
+        self.assertEqual(self.request('GET','/api/v1/reports/activity?from=2020-01-01&to=2035-01-01')[0],400)
+
+if __name__ == '__main__': unittest.main()