#!/usr/bin/env python3 """ Phase 4: Kanban Pipeline Board (MissionPacket-backed) Every card is a MissionPacket.v1 — a durable packet with scope, budget, expiry, denied actions, approval requirements, and content hash. 9 columns with entry gates, WIP limits, and exit conditions: 1. Universe — gallery exists, any state WIP ∞ → ATHENA promotes 2. Qualified — trading_status CERTIFIED + comp model known + degree bar known WIP 25 3. Prepared — current-show hook CERTIFIED + one artist brief drilled to 90% + opener written WIP 8 4. Visited — ≥1 visit observation in H3 ledger WIP 8 5. Contact — named human captured + follow-up sent (human-approved) WIP 5 6. Applied — application submitted (human action) WIP 5 7. Interview — scheduled WIP 3 8. Offer — terms in hand WIP ∞ 9. Closed — declined/rejected/expired, with reason WIP ∞ terminal WIP limits are HARD. ATHENA may not promote into a full column; must surface which current card to close. Seeded from §6.3 of the requirements doc: Qualified → Prepared candidates (in priority order): 1. Findlay Galleries — "substantial commissions and bonus plans" — education preferred not required 2. LIK Fine Art — "true six-figure income potential" — no degree stated 3. EDEN House of Art — "Competitive base salary plus commission structure" — no degree stated 4. Avant Gallery — 7-day retail hours — no live posting — unknown comp 5. Martin Lawrence — 51–200 employees — unknown comp 6. VFA — "$25 hourly plus commission" — 3 yrs art industry 7. Sorrel Sky — "$90-130K annually based on sales" — BA required 8. Taglialatella — no live posting — unknown comp Closed at seed (with reason): - Mnuchin (ceased trading Feb 2026) - Twelve Chairs (commission-only, no base — unviable at 18) - Cavalier / West Chelsea / Contessa (non-NYC HQ or dormant) """ import hashlib import json import sqlite3 import sys import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from certification_spine import ( CertificationEngine, CertificationState, DEFAULT_DB as CERT_DB, ) PIPELINE_DB = Path(__file__).parent / "avery_pipeline.db" # ── MissionPacket ────────────────────────────────────────────────────── @dataclass class MissionPacket: """A durable work unit. Every pipeline card is a packet.""" id: str # packet_id gallery_id: str # gallery canonical ID gallery_name: str # human-readable scope: str # what this packet covers budget: str # time budget expiry: str # ISO 8601 UTC denied_actions: list[str] # things this packet forbids approval_required: bool = True # C2 — always True for comms content_hash: str = "" # SHA-256 of the canonical card state created_at: str = "" # ISO 8601 status: str = "open" # open | claimed | in_progress | completed | closed claimed_by: str = "" # profile ID that claimed it (exactly-once) column: str = "Universe" # current pipeline column wip_slot: int = 0 # position within WIP-limited column reason_closed: str = "" # why it left the pipeline capsule_id: str = "avery-nyc-gallery" updated_at: str = "" @dataclass class PipelineCard: packet: MissionPacket priority: int = 0 # seed priority (1 = highest) drill_score: float = 0.0 # 0–100, from COACH opener: str = "" # CLOSER draft, human-approved current_show_hook: str = "" # from CURATOR artist_brief: str = "" # from CURATOR visit_observations: list[str] = field(default_factory=list) named_human: str = "" # captured contact comp_model: str = "" # comp language from spine degree_bar: str = "" # education requirement certifiable_comp: bool = False # comp reaches CERTIFIED state? # ── Pipeline Store ───────────────────────────────────────────────────── class PipelineStore: """SQLite-backed pipeline. WIP enforcement, exactly-once claim, column gates.""" WIP_LIMITS = { "Universe": float('inf'), "Qualified": 25, "Prepared": 8, "Visited": 8, "Contact": 5, "Applied": 5, "Interview": 3, "Offer": float('inf'), "Closed": float('inf'), } ENTRY_GATES = { "Universe": lambda c: True, "Qualified": lambda c: ( c.certifiable_comp and bool(c.comp_model) and bool(c.degree_bar) ), "Prepared": lambda c: ( bool(c.current_show_hook) and c.drill_score >= 90.0 and bool(c.opener) ), "Visited": lambda c: len(c.visit_observations) >= 1, "Contact": lambda c: bool(c.named_human), "Applied": lambda c: True, # human action — not enforced by system "Interview": lambda c: True, # human action — not enforced by system "Offer": lambda c: True, "Closed": lambda c: True, } def __init__(self, db: Path): self.db = db self._ensure_schema() def _ensure_schema(self): conn = sqlite3.connect(self.db) conn.execute(""" CREATE TABLE IF NOT EXISTS packets ( packet_id TEXT PRIMARY KEY, capsule_id TEXT, gallery_id TEXT NOT NULL, gallery_name TEXT NOT NULL, scope TEXT, budget TEXT, expiry TEXT, denied_actions TEXT, -- JSON array approval_required INTEGER DEFAULT 1, content_hash TEXT, created_at TEXT, status TEXT DEFAULT 'open', claimed_by TEXT DEFAULT '', column TEXT DEFAULT 'Universe', wip_slot INTEGER DEFAULT 0, reason_closed TEXT DEFAULT '', updated_at TEXT ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS card_faces ( packet_id TEXT PRIMARY KEY REFERENCES packets(packet_id), priority INTEGER DEFAULT 0, drill_score REAL DEFAULT 0, opener TEXT DEFAULT '', current_show_hook TEXT DEFAULT '', artist_brief TEXT DEFAULT '', visit_obs TEXT DEFAULT '[]', -- JSON array named_human TEXT DEFAULT '', comp_model TEXT DEFAULT '', degree_bar TEXT DEFAULT '', certifiable_comp INTEGER DEFAULT 0, updated_at TEXT ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_packets_column ON packets(column, wip_slot)") conn.execute("CREATE INDEX IF NOT EXISTS idx_packets_status ON packets(status)") conn.commit() conn.close() # ── packet ops ───────────────────────────────────────────────────── def create_packet(self, card: PipelineCard) -> MissionPacket: now = datetime.now(timezone.utc).isoformat() pid = f"mp_{uuid.uuid4().hex[:12]}" denied = json.dumps(card.packet.denied_actions) face_data = json.dumps(card.visit_observations) conn = sqlite3.connect(self.db) # Content hash over the canonical state canonical = json.dumps({ "gallery_id": card.packet.gallery_id, "gallery_name": card.packet.gallery_name, "column": card.packet.column, "scope": card.packet.scope, "priority": card.priority, "comp_model": card.comp_model, "degree_bar": card.degree_bar, "created_at": now, }, sort_keys=True) chash = hashlib.sha256(canonical.encode()).hexdigest() conn.execute( """INSERT INTO packets VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( pid, card.packet.capsule_id, card.packet.gallery_id, card.packet.gallery_name, card.packet.scope, card.packet.budget, card.packet.expiry, denied, int(card.packet.approval_required), chash, now, card.packet.status, card.packet.claimed_by, card.packet.column, card.packet.wip_slot, card.packet.reason_closed, now, ), ) conn.execute( """INSERT INTO card_faces VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", ( pid, card.priority, card.drill_score, card.opener, card.current_show_hook, card.artist_brief, face_data, card.named_human, card.comp_model, card.degree_bar, int(card.certifiable_comp), now, ), ) conn.commit() conn.close() card.packet.id = pid card.packet.content_hash = chash card.packet.created_at = now print(f" Created packet {pid}: {card.packet.gallery_name} → {card.packet.column} (priority {card.priority})") return card.packet def claim_packet(self, packet_id: str, profile_id: str) -> bool: """ Exactly-once claim. Returns True if claim succeeded. A packet can only be claimed once — prevents two profiles from both 'working' the same gallery (Requirement 6.1). """ conn = sqlite3.connect(self.db) cursor = conn.execute("SELECT claimed_by, status FROM packets WHERE packet_id=?", (packet_id,)) row = cursor.fetchone() if not row: conn.close() return False claimed_by, status = row if claimed_by: conn.close() print(f" ✗ Claim failed: {packet_id} already claimed by {claimed_by}") return False if status not in ("open",): conn.close() print(f" ✗ Claim failed: {packet_id} status is {status}") return False now = datetime.now(timezone.utc).isoformat() conn.execute( "UPDATE packets SET status='claimed', claimed_by=?, updated_at=? WHERE packet_id=?", (profile_id, now, packet_id), ) conn.commit() conn.close() print(f" ✓ {packet_id} claimed by {profile_id}") return True def get_card(self, packet_id: str) -> PipelineCard | None: conn = sqlite3.connect(self.db) cursor = conn.execute("SELECT * FROM packets WHERE packet_id=?", (packet_id,)) p_row = cursor.fetchone() if not p_row: conn.close() return None cursor = conn.execute("SELECT * FROM card_faces WHERE packet_id=?", (packet_id,)) f_row = cursor.fetchone() conn.close() pk = MissionPacket( id=p_row[0], capsule_id=p_row[1], gallery_id=p_row[2], gallery_name=p_row[3], scope=p_row[4], budget=p_row[5], expiry=p_row[6], denied_actions=json.loads(p_row[7] or '[]'), approval_required=bool(p_row[8]), content_hash=p_row[9], created_at=p_row[10], status=p_row[11], claimed_by=p_row[12], column=p_row[13], wip_slot=p_row[14], reason_closed=p_row[15], updated_at=p_row[16], ) face = { "priority": f_row[1], "drill_score": f_row[2], "opener": f_row[3], "current_show_hook": f_row[4], "artist_brief": f_row[5], "visit_observations": json.loads(f_row[6] or '[]'), "named_human": f_row[7], "comp_model": f_row[8], "degree_bar": f_row[9], "certifiable_comp": bool(f_row[10]), } return PipelineCard(packet=pk, **face) def column_count(self, column: str) -> int: conn = sqlite3.connect(self.db) cursor = conn.execute( "SELECT COUNT(*) FROM packets WHERE column=? AND status IN ('open','claimed','in_progress')", (column,), ) count = cursor.fetchone()[0] conn.close() return count def can_promote(self, packet_id: str, target_column: str) -> tuple[bool, str]: """ Check if a card can be promoted to target_column. Returns (allowed, reason). """ card = self.get_card(packet_id) if not card: return False, "packet not found" # WIP limit check current_count = self.column_count(target_column) wip = self.WIP_LIMITS[target_column] if current_count >= wip: return False, ( f"WIP limit reached for {target_column}: {current_count}/{wip} slots filled. " f"ATHENA must surface which existing card to close before promoting." ) # Entry gate check gate = self.ENTRY_GATES[target_column] if not gate(card): missing = [] if target_column == "Qualified": if not card.certifiable_comp: missing.append("comp not certifiable (T4 vendor estimate — ineligible)") if not card.comp_model: missing.append("no comp model captured") if not card.degree_bar: missing.append("no degree bar captured") if target_column == "Prepared": if not card.current_show_hook: missing.append("no current-show hook (CURATOR)") if card.drill_score < 90.0: missing.append(f"drill score {card.drill_score}% — below 90% gate (COACH)") if not card.opener: missing.append("no opener written (CLOSER draft, human-approved)") if target_column == "Visited": if len(card.visit_observations) < 1: missing.append("no visit observation in H3 ledger (CARTOGRAPHER)") if target_column == "Contact": if not card.named_human: missing.append("no named human captured") return False, f"Entry gate for {target_column} not met: {'; '.join(missing)}" return True, "ok" def promote(self, packet_id: str, target_column: str, promoted_by: str) -> tuple[bool, str]: """ Promote a card to target_column. Runs can_promote first. Returns (success, message). """ allowed, reason = self.can_promote(packet_id, target_column) if not allowed: return False, reason conn = sqlite3.connect(self.db) now = datetime.now(timezone.utc).isoformat() # Find next WIP slot cursor = conn.execute( "SELECT COALESCE(MAX(wip_slot),0) FROM packets WHERE column=?", (target_column,), ) next_slot = cursor.fetchone()[0] + 1 conn.execute( "UPDATE packets SET column=?, wip_slot=?, updated_at=? WHERE packet_id=?", (target_column, next_slot, now, packet_id), ) # If card was in a WIP-limited column, free the slot old_card = self.get_card(packet_id) conn.close() print(f" ✓ {packet_id} ({old_card.packet.gallery_name}) → {target_column} (slot {next_slot}) promoted by {promoted_by}") return True, "promoted" def close_card(self, packet_id: str, reason: str, closed_by: str) -> bool: """ Move a card to Closed column with a reason. """ conn = sqlite3.connect(self.db) now = datetime.now(timezone.utc).isoformat() conn.execute( "UPDATE packets SET column='Closed', reason_closed=?, status='completed', updated_at=? WHERE packet_id=?", (reason, now, packet_id), ) conn.commit() conn.close() print(f" ⊘ {packet_id} closed — {reason} (closed by {closed_by})") return True def list_cards(self, columns: list[str] | None = None) -> list[PipelineCard]: conn = sqlite3.connect(self.db) if columns: placeholders = ','.join('?' * len(columns)) cursor = conn.execute( f"SELECT packet_id FROM packets WHERE column IN ({placeholders}) ORDER BY wip_slot", columns, ) rows = cursor.fetchall() else: cursor = conn.execute("SELECT packet_id FROM packets ORDER BY CASE column " + "WHEN 'Universe' THEN 1 WHEN 'Qualified' THEN 2 WHEN 'Prepared' THEN 3 " + "WHEN 'Visited' THEN 4 WHEN 'Contact' THEN 5 WHEN 'Applied' THEN 6 " + "WHEN 'Interview' THEN 7 WHEN 'Offer' THEN 8 WHEN 'Closed' THEN 9 END, wip_slot") rows = cursor.fetchall() conn.close() return [self.get_card(r[0]) for r in rows if r[0]] def print_board(self): """Print the full board in column order.""" columns = ["Universe", "Qualified", "Prepared", "Visited", "Contact", "Applied", "Interview", "Offer", "Closed"] print(f"\n{'='*70}") print(f"KANBAN BOARD — ATHENA orchestrator · {self.db.name}") print(f"{'='*70}") for col in columns: cards = [c for c in self.list_cards([col]) if c] wip = self.WIP_LIMITS[col] suffix = f" (WIP {len(cards)}/{wip})" if wip != float('inf') else f" (WIP ∞)" print(f"\n 📋 {col}{suffix}") if not cards: print(f" (empty)") continue for c in cards: pk = c.packet face = c status_mark = "●" if pk.status == "claimed" else ("◉" if pk.status == "in_progress" else "○") comp_mark = "✓ comp" if face.certifiable_comp else "(T4 — unverifiable)" drill = f"drill {face.drill_score:.0f}%" if face.drill_score > 0 else "" print(f" {status_mark} [#{pk.wip_slot}] {pk.gallery_name}") print(f" priority={face.priority} · {comp_mark} · {drill}") if face.current_show_hook: print(f" show: {face.current_show_hook[:70]}") if face.named_human: print(f" contact: {face.named_human}") if pk.reason_closed: print(f" CLOSED: {pk.reason_closed}") print() # ── Seed data from §6.3 ──────────────────────────────────────────────── SEED_CARDS = [ # priority, gallery_id, gallery_name, comp_model, degree_bar, certifiable_comp (1, "findlay", "Findlay Galleries — 32 E 57th St", '"substantial commissions and bonus plans"', "Education preferred, not required", True), (2, "lik", "LIK Fine Art — 689 Fifth Avenue", '"true six-figure income potential"', "None stated; 3 yrs sales preferred", True), (3, "eden", "EDEN House of Art — SoHo + Fifth Ave", '"Competitive base salary plus commission structure on art sales"', "None stated; 2 yrs luxury direct sales preferred", True), (4, "avant", "Avant Gallery — 20 Hudson Yards", "n.a. — no live posting", "Unknown", False), (5, "martin_lawrence","Martin Lawrence Galleries — SoHo", "n.a.; 51–200 employees (Indeed)", "Unknown", False), (6, "vf_a", "VFA — NYC", '"$25 hourly plus commission"', "3 yrs art industry; no education stated", True), (7, "sorrel_sky", "Sorrel Sky Gallery — SoHo", '"$90-130K annually based on sales"', "BA in art/business; min 2 yrs", True), (8, "taglialatella", "Taglialatella Galleries — 229 10th Ave, Chelsea", "n.a. — no live posting", "Unknown", False), ] def seed_board(store: PipelineStore): """Seed the pipeline board. Clears existing data first.""" # Clear existing data to avoid duplicates from re-runs conn = sqlite3.connect(store.db) conn.execute("DELETE FROM card_faces") conn.execute("DELETE FROM packets") conn.commit() conn.close() print("Seeding board from §6.3 of requirements doc...\n") for priority, gid, name, comp, degree, cert_comp in SEED_CARDS: now = datetime.now(timezone.utc).isoformat() exp = (datetime.now(timezone.utc).replace(year=datetime.now(timezone.utc).year + 1) .isoformat()) packet = MissionPacket( id="", capsule_id="avery-nyc-gallery", gallery_id=gid, gallery_name=name, scope=f"Target {name} for commissioned art-sales role", budget="2 weeks active pursuit", expiry=exp, denied_actions=["send_email", "submit_application_autonomously", "contact_gallery_without_approval"], approval_required=True, content_hash="", created_at=now, status="open", column="Universe", ) card = PipelineCard( packet=packet, priority=priority, comp_model=comp, degree_bar=degree, certifiable_comp=cert_comp, current_show_hook="", artist_brief="", opener="", named_human="", visit_observations=[], drill_score=0.0, ) store.create_packet(card) # Closed at seed closed_seed = [ ("mnuchin", "Closed — ceased trading February 2026 after death of owner Robert Mnuchin (Dec 20, 2025)", "ATHENA"), ("twelve_chairs", "Closed — commission-only, no base; unviable at age 18", "ATHENA"), ("cavalier", "Closed — non-NYC HQ (Greenwich CT), dormant", "ATHENA"), ("west_chelsea", "Closed — HQ Austin TX; NY presence limited; dormant", "ATHENA"), ("contessa", "Closed — Hamptons-season model, not year-round Manhattan", "ATHENA"), ] for gid, reason, by in closed_seed: now = datetime.now(timezone.utc).isoformat() packet = MissionPacket( id="", capsule_id="avery-nyc-gallery", gallery_id=gid, gallery_name=gid, scope="Seed — closed", budget="", expiry=now, denied_actions=[], approval_required=False, content_hash="", created_at=now, status="completed", column="Closed", reason_closed=reason, ) card = PipelineCard(packet=packet, priority=99) store.create_packet(card) store.print_board() if __name__ == "__main__": print("╔══════════════════════════════════════════════════════════════╗") print("║ Phase 4: Kanban Pipeline Board (MissionPacket-backed) ║") print("║ AverysArtJourney · 9 columns · WIP enforcement · seeded ║") print("╚══════════════════════════════════════════════════════════════╝") store = PipelineStore(PIPELINE_DB) seed_board(store) print(" ✓ Phase 4 complete — seed board built") print(f" DB: {PIPELINE_DB}")