# P1: Sales Pipeline Bridge and Routing Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Make a pipeline stage change create a correctly routed Hermes kanban task, recorded idempotently in a `deal_map` table.

**Architecture:** n8n signs one HMAC-SHA256 V2 POST per stage change to a single Hermes orchestrator webhook route (`deal-stage`). The orchestrator reads a stage-to-domain map and creates a kanban task. A deterministic plugin records the deal-to-task link in `deal_map`. A verification script gates task closure.

**Tech Stack:** Hermes Agent (profiles, kanban, plugins, native webhooks), bash, Python 3, Postgres (psql), n8n (JSON workflows, n8n-mcp), YAML.

**Spec:** `../specs/2026-09-24-sales-pipeline-hermes-design.md` (SP1, sections 4, 5.1, 9, 11, 12).

---

## Prerequisites

This plan edits the fleet repo, not this content repo. Before starting:

- [ ] **Clone the fleet repo and check out the branch**
  ```bash
  git clone git@github.com:Coordenadas-co/factronics-hermes-fleet.git
  cd factronics-hermes-fleet
  git checkout audit/icm-contract-fidelity
  ```
- [ ] **Tooling available:** `bash`, `python3` (with `pyyaml`), `psql`, `git`, and an n8n-mcp connection to the live n8n instance (`n8n.916.coordenadas.co`).
- [ ] **Env for local runs:** `DATABASE_URL` pointing at the shared fleet Postgres.
- [ ] **Discovery (one time):** confirm the exact kanban lifecycle hook payload so the plugin can read deal metadata. Run:
  ```bash
  python3 - <<'PY'
  import inspect, hermes_cli.kanban_db as k
  print(inspect.getsource(k._fire_kanban_lifecycle_hook))
  PY
  ```

  Expected: the hook passes `task_id`, `profile_name`, and the task row (including `metadata`). Record the exact key that carries `metadata`. If metadata is NOT present, stop and report; Task 4 has a fallback note.

All commands below run from the fleet repo root unless stated otherwise.

---

## File Structure


| Path                                                | Responsibility                                          |
| --------------------------------------------------- | ------------------------------------------------------- |
| `workspace/shared/deal_stages.yaml`                 | Canonical stage to (domain, task\_type) map             |
| `workspace/shared/deal_map.sql`                     | `deal_map` table DDL plus idempotent migration          |
| `workspace/scripts/verify_deal_map.sh`              | Gate: a task exists for (deal\_id, dealstage)           |
| `plugins/kanban-hubspot-bridge/__init__.py`         | Deterministic plugin: UPSERT `deal_map` on task claim   |
| `plugins/kanban-hubspot-bridge/README.md`           | Plugin doc                                              |
| `scripts/render-deal-stage-route.py`                | Render the webhook route prompt from `deal_stages.yaml` |
| `profiles/orchestrator/config.yaml`                 | Add the `deal-stage` webhook route                      |
| `n8n/factronics-forms-live/webhook-deal-stage.json` | Connect outputs, add the signing branch                 |
| `AGENTS.md`                                         | Routing table update                                    |
| `n8n/README.md`                                     | Workflow status update                                  |


---

## Task 1: Canonical routing map

**Files:**

- Create: `workspace/shared/deal_stages.yaml`
- Test: `workspace/shared/test_deal_stages.py`

- [ ] **Step 1: Write the failing test**

Create `workspace/shared/test_deal_stages.py`:

```python
import pathlib
import yaml

HERE = pathlib.Path(__file__).parent
ALLOWED_DOMAINS = {"orchestrator", "contable", "inbox", "operations", "onboarding"}
EXPECTED_STAGES = {
    "appointmentscheduled": ("inbox", "deal_qualified"),
    "qualifiedtobuy": ("operations", "deal_product_details_track"),
    "presentationscheduled": ("orchestrator", "deal_presentation_prep"),
    "decisionmakerboughtin": ("inbox", "deal_negotiation_monitor"),
    "contractsent": ("operations", "deal_close_expectation"),
    "closedwon": ("onboarding", "onboarding_intake"),
    "closedlost": ("operations", "deal_lost_record"),
}


def load():
    return yaml.safe_load((HERE / "deal_stages.yaml").read_text())


def test_all_seven_stages_present():
    data = load()
    assert set(data["stages"].keys()) == set(EXPECTED_STAGES.keys())


def test_domain_and_task_type():
    data = load()
    for stage, (domain, task_type) in EXPECTED_STAGES.items():
        entry = data["stages"][stage]
        assert entry["domain"] == domain, stage
        assert entry["task_type"] == task_type, stage


def test_domains_are_within_the_fleet():
    data = load()
    for stage, entry in data["stages"].items():
        assert entry["domain"] in ALLOWED_DOMAINS, stage


def test_no_gateway_domain():
    data = load()
    for entry in data["stages"].values():
        assert entry["domain"] != "gateway"
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest workspace/shared/test_deal_stages.py -v`
Expected: FAIL with `FileNotFoundError` for `deal_stages.yaml`.

- [ ] **Step 3: Create the routing map**

Create `workspace/shared/deal_stages.yaml`:

```yaml
# deal_stages.yaml — canonical stage -> (domain, task_type) routing for the sales pipeline bridge.
# Single source of truth. Rendered into the orchestrator webhook route by
# scripts/render-deal-stage-route.py. Do not hardcode this map anywhere else.
version: 1
stages:
  appointmentscheduled:
    name: "Calificar"
    domain: inbox
    task_type: deal_qualified
  qualifiedtobuy:
    name: "Proyecto Viable"
    domain: operations
    task_type: deal_product_details_track
  presentationscheduled:
    name: "Presentacion de Solucion"
    domain: orchestrator
    task_type: deal_presentation_prep
  decisionmakerboughtin:
    name: "Negociacion/Revision"
    domain: inbox
    task_type: deal_negotiation_monitor
  contractsent:
    name: "Expectativa de Cierre"
    domain: operations
    task_type: deal_close_expectation
  closedwon:
    name: "Ganado"
    domain: onboarding
    task_type: onboarding_intake
  closedlost:
    name: "Perdido"
    domain: operations
    task_type: deal_lost_record
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest workspace/shared/test_deal_stages.py -v`
Expected: 4 passed.

- [ ] **Step 5: Commit**

```bash
git add workspace/shared/deal_stages.yaml workspace/shared/test_deal_stages.py
git commit -m "feat(pipeline): canonical deal stage routing map"
```

---

## Task 2: deal\_map table

**Files:**

- Create: `workspace/shared/deal_map.sql`
- Test: `workspace/shared/test_deal_map_sql.sh`

- [ ] **Step 1: Write the failing test**

Create `workspace/shared/test_deal_map_sql.sh`:

```bash
#!/usr/bin/env bash
set -euo pipefail
here="$(cd "$(dirname "$0")" && pwd)"
sql="$here/deal_map.sql"

if [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:document-link:!%20-f%20%22%24sql%22]]; then echo "FAIL: deal_map.sql missing"; exit 1; fi
if ! grep -q "PRIMARY KEY (deal_id, dealstage)" "$sql"; then
  echo "FAIL: missing composite primary key"; exit 1
fi
if ! grep -qi "IF NOT EXISTS" "$sql"; then echo "FAIL: not idempotent"; exit 1; fi

if [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:document-link:-n%20%22%24%7BDATABASE_URL%3A-%7D%22]] && command -v psql >/dev/null 2>&1; then
  psql "$DATABASE_URL" -v ON_ERROR_STOP=1 -f "$sql" >/dev/null
  psql "$DATABASE_URL" -v ON_ERROR_STOP=1 -f "$sql" >/dev/null   # idempotent second run
  cols=$(psql "$DATABASE_URL" -tA -c \
    "SELECT count(*) FROM information_schema.columns WHERE table_name='deal_map';")
  [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:document-link:%22%24cols%22%20%3D%3D%20%227%22]] || { echo "FAIL: expected 7 columns, got $cols"; exit 1; }
fi
echo "OK"
```

- [ ] **Step 2: Run test to verify it fails**

Run: `bash workspace/shared/test_deal_map_sql.sh`
Expected: `FAIL: deal_map.sql missing` (exit 1).

- [ ] **Step 3: Create the schema**

Create `workspace/shared/deal_map.sql`:

```sql
-- deal_map — same Postgres as audit_log. Links a (deal, stage) transition to the Hermes
-- kanban task it produced. Idempotent by primary key, mirroring kill_switch.sql.
CREATE TABLE IF NOT EXISTS deal_map (
  deal_id        text        NOT NULL,
  dealstage      text        NOT NULL,
  kanban_task_id text,
  domain         text,
  task_type      text,
  created_at     timestamptz NOT NULL DEFAULT now(),
  updated_at     timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (deal_id, dealstage)
);

CREATE INDEX IF NOT EXISTS deal_map_task_idx ON deal_map (kanban_task_id);

-- Upsert used by the kanban-hubspot-bridge plugin (task id may arrive after first insert):
--   INSERT INTO deal_map (deal_id, dealstage, kanban_task_id, domain, task_type)
--   VALUES ($1, $2, $3, $4, $5)
--   ON CONFLICT (deal_id, dealstage) DO UPDATE
--     SET kanban_task_id = EXCLUDED.kanban_task_id,
--         domain = EXCLUDED.domain, task_type = EXCLUDED.task_type, updated_at = now();
```

- [ ] **Step 4: Run test to verify it passes**

Run: `bash workspace/shared/test_deal_map_sql.sh`
Expected: `OK` (with `DATABASE_URL` set it also applies twice and checks 7 columns).

- [ ] **Step 5: Commit**

```bash
git add workspace/shared/deal_map.sql workspace/shared/test_deal_map_sql.sh
git commit -m "feat(pipeline): deal_map schema"
```

---

## Task 3: verify\_deal\_map.sh

**Files:**

- Create: `workspace/scripts/verify_deal_map.sh`
- Test: `workspace/scripts/test_verify_deal_map.sh`

- [ ] **Step 1: Write the failing test**

Create `workspace/scripts/test_verify_deal_map.sh`:

```bash
#!/usr/bin/env bash
set -uo pipefail
here="$(cd "$(dirname "$0")" && pwd)"
script="$here/verify_deal_map.sh"
fails=0

# No args -> usage, exit 2
"$script" >/dev/null 2>&1; [[ $? -eq 2 ]] || { echo "FAIL: no-args should exit 2"; fails=1; }

# No DATABASE_URL -> BLOCKED, exit 1
out=$(env -u DATABASE_URL "$script" d1 appointmentscheduled 2>&1); code=$?
[[ $code -eq 1 ]] || { echo "FAIL: empty DATABASE_URL should exit 1"; fails=1; }
echo "$out" | grep -q "BLOCKED" || { echo "FAIL: expected BLOCKED message"; fails=1; }

# Missing rows -> NOT FOUND, exit 1 (only runs online)
if [[ -n "${DATABASE_URL:-}" ]] && command -v psql >/dev/null 2>&1; then
  "$script" __nope__ __nope__ >/dev/null 2>&1; [[ $? -eq 1 ]] || { echo "FAIL: missing row should exit 1"; fails=1; }
fi

[[ $fails -eq 0 ]] && echo "OK" || exit 1
```

- [ ] **Step 2: Run test to verify it fails**

Run: `bash workspace/scripts/test_verify_deal_map.sh`
Expected: FAIL (script missing).

- [ ] **Step 3: Write the script**

Create `workspace/scripts/verify_deal_map.sh`:

```bash
#!/usr/bin/env bash
# verify_deal_map.sh [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:inline-html:%3Cdeal_id%3E]] [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:inline-html:%3Cdealstage%3E]]
#
# Gate: confirm the (deal_id, dealstage) transition produced a kanban task.
# Same fail-closed rule as verify_audit.sh: if the check tool is unavailable, BLOCK.
set -euo pipefail

deal_id="${1:-}"
dealstage="${2:-}"

if [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:document-link:-z%20%22%24deal_id%22%7C%7C%20-z%20%22%24dealstage%22]]; then
  echo "uso: verify_deal_map.sh [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:inline-html:%3Cdeal_id%3E]] [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:inline-html:%3Cdealstage%3E]]" >&2
  exit 2
fi

if [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:document-link:-z%20%22%24%7BDATABASE_URL%3A-%7D%22]]; then
  echo "BLOCKED: DATABASE_URL no esta seteado — no se asume exito" >&2
  exit 1
fi

if ! command -v psql >/dev/null 2>&1; then
  echo "BLOCKED: psql no esta instalado — no se asume exito" >&2
  exit 1
fi

result=$(psql "$DATABASE_URL" -tA -c \
  "SELECT 1 FROM deal_map WHERE deal_id = '${deal_id//\'/\'\'}' AND dealstage = '${dealstage//\'/\'\'}' AND kanban_task_id IS NOT NULL LIMIT 1;" \
  2>/dev/null || true)

if [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:document-link:%22%24result%22%20%3D%3D%20%221%22]]; then
  exit 0
fi

echo "NOT FOUND: sin task en deal_map para deal_id=${deal_id} dealstage=${dealstage}" >&2
exit 1
```

- [ ] **Step 4: Run test to verify it passes**

Run: `bash workspace/scripts/test_verify_deal_map.sh`
Expected: `OK`.

- [ ] **Step 5: Make executable and commit**

```bash
chmod +x workspace/scripts/verify_deal_map.sh workspace/scripts/test_verify_deal_map.sh
git add workspace/scripts/verify_deal_map.sh workspace/scripts/test_verify_deal_map.sh
git commit -m "feat(pipeline): verify_deal_map gate"
```

---

## Task 4: kanban-hubspot-bridge plugin

**Files:**

- Create: `plugins/kanban-hubspot-bridge/__init__.py`
- Create: `plugins/kanban-hubspot-bridge/README.md`
- Test: `plugins/kanban-hubspot-bridge/test_bridge.py`

- [ ] **Step 1: Write the failing test**

Create `plugins/kanban-hubspot-bridge/test_bridge.py`:

```python
import importlib.util
import pathlib
import sqlite3

HERE = pathlib.Path(__file__).parent


def load_plugin():
    spec = importlib.util.spec_from_file_location("bridge", HERE / "__init__.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


class FakeCtx:
    def __init__(self):
        self.hooks = {}

    def register_hook(self, name, fn):
        self.hooks[name] = fn


def _db_with_task(tmp_path, metadata_json):
    db = tmp_path / "kanban.db"
    con = sqlite3.connect(db)
    con.execute("CREATE TABLE tasks (id TEXT PRIMARY KEY, metadata TEXT, assignee TEXT)")
    con.execute("INSERT INTO tasks (id, metadata, assignee) VALUES (?,?,?)",
                ("t_1", metadata_json, "contable"))
    con.commit()
    con.close()
    return db


def test_extracts_deal_metadata(tmp_path, monkeypatch):
    mod = load_plugin()
    db = _db_with_task(tmp_path, '{"deal_id": "D123", "dealstage": "closedwon"}')
    monkeypatch.setenv("KANBAN_DB", str(db))
    meta = mod.read_deal_metadata("t_1")
    assert meta == {"deal_id": "D123", "dealstage": "closedwon"}


def test_upsert_sql_shape():
    mod = load_plugin()
    sql = mod.UPSERT_SQL.lower()
    assert "insert into deal_map" in sql
    assert "on conflict (deal_id, dealstage)" in sql


def test_no_deal_id_is_noop(tmp_path, monkeypatch):
    mod = load_plugin()
    db = _db_with_task(tmp_path, '{}')
    monkeypatch.setenv("KANBAN_DB", str(db))
    called = []
    monkeypatch.setattr(mod, "_upsert", lambda *a, **k: called.append(a))
    ctx = FakeCtx()
    mod.register(ctx)
    ctx.hooks["kanban_task_claimed"](task_id="t_1", profile_name="contable")
    assert called == []
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest plugins/kanban-hubspot-bridge/test_bridge.py -v`
Expected: FAIL (plugin missing).

- [ ] **Step 3: Write the plugin**

Create `plugins/kanban-hubspot-bridge/__init__.py`:

```python
"""kanban-hubspot-bridge — records the deal<->task link in deal_map.

Deterministic, no LLM. When a task whose metadata carries a deal_id is claimed,
UPSERT a row into deal_map. Best-effort: without DATABASE_URL or psql the plugin
no-ops and never blocks the kanban lifecycle (mirrors kanban-asana-bridge).
"""
import json
import os
import sqlite3
import subprocess

UPSERT_SQL = (
    "INSERT INTO deal_map (deal_id, dealstage, kanban_task_id, domain, task_type) "
    "VALUES ('{deal_id}', '{dealstage}', '{task_id}', '{domain}', '{task_type}') "
    "ON CONFLICT (deal_id, dealstage) DO UPDATE "
    "SET kanban_task_id = EXCLUDED.kanban_task_id, "
    "domain = EXCLUDED.domain, task_type = EXCLUDED.task_type, updated_at = now();"
)


def _sql_escape(value: str) -> str:
    return str(value).replace("'", "''")


def read_deal_metadata(task_id: str) -> dict:
    """Read a task's metadata JSON from the kanban SQLite DB. Fail-open to {}."""
    db_path = os.environ.get("KANBAN_DB")
    if not db_path or not os.path.exists(db_path):
        return {}
    try:
        con = sqlite3.connect(db_path)
        row = con.execute("SELECT metadata FROM tasks WHERE id = ?", (task_id,)).fetchone()
        con.close()
        if not row or not row[0]:
            return {}
        data = json.loads(row[0])
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _upsert(deal_id, dealstage, task_id, domain, task_type) -> None:
    dsn = os.environ.get("DATABASE_URL")
    if not dsn:
        return
    sql = UPSERT_SQL.format(
        deal_id=_sql_escape(deal_id),
        dealstage=_sql_escape(dealstage),
        task_id=_sql_escape(task_id),
        domain=_sql_escape(domain),
        task_type=_sql_escape(task_type),
    )
    try:
        subprocess.run(["psql", dsn, "-v", "ON_ERROR_STOP=1", "-c", sql],
                       check=True, capture_output=True, timeout=10)
    except Exception:
        pass  # best-effort


def register(ctx):
    def on_claimed(task_id=None, profile_name=None, **kw):
        meta = read_deal_metadata(task_id) if task_id else {}
        deal_id = meta.get("deal_id")
        dealstage = meta.get("dealstage")
        if not deal_id or not dealstage:
            return  # nothing to map
        _upsert(deal_id, dealstage, task_id,
                meta.get("domain", profile_name or ""), meta.get("task_type", ""))

    ctx.register_hook("kanban_task_claimed", on_claimed)
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest plugins/kanban-hubspot-bridge/test_bridge.py -v`
Expected: 3 passed.

- [ ] **Step 5: Write the plugin README**

Create `plugins/kanban-hubspot-bridge/README.md`:

```markdown
# kanban-hubspot-bridge

Hermes plugin (`kind: standalone`), zero LLM. On `kanban_task_claimed`, if the task
metadata carries `deal_id` and `dealstage`, UPSERT a `deal_map` row (deal -> task).

## Config

- `DATABASE_URL` — shared fleet Postgres. Unset -> no-op (best-effort).
- `KANBAN_DB` — path to the kanban SQLite file, used to read task metadata.

## Install

`provision.sh` copies this plugin into `orchestrator` (the dispatcher process sees the
`claimed` hook). Requires `psql` on PATH; without it the plugin no-ops.

## Note

If the kanban lifecycle hook does not pass a task metadata column readable here, the
fallback is for the orchestrator to encode `deal_id` in the task title
(`[deal:<id>] ...`) and this plugin to parse it. Confirm during the Prerequisites
discovery step.
```

- [ ] **Step 6: Commit**

```bash
git add plugins/kanban-hubspot-bridge/
git commit -m "feat(pipeline): kanban-hubspot-bridge plugin writes deal_map"
```

---

## Task 5: Orchestrator webhook route

**Files:**

- Create: `scripts/render-deal-stage-route.py`
- Test: `scripts/test_render_deal_stage_route.py`
- Modify: `profiles/orchestrator/config.yaml`

- [ ] **Step 1: Write the failing test**

Create `scripts/test_render_deal_stage_route.py`:

```python
import importlib.util
import pathlib

HERE = pathlib.Path(__file__).parent
ROOT = HERE.parent


def load():
    spec = importlib.util.spec_from_file_location("r", HERE / "render-deal-stage-route.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def test_renders_every_stage():
    mod = load()
    text = mod.render()
    for stage in ["appointmentscheduled", "qualifiedtobuy", "presentationscheduled",
                  "decisionmakerboughtin", "contractsent", "closedwon", "closedlost"]:
        assert stage in text, stage


def test_mentions_no_terminal_tools():
    mod = load()
    text = mod.render()
    assert "terminal" not in text and "file" not in text and "code" not in text


def test_config_matches_render():
    mod = load()
    rendered = mod.render().strip()
    cfg = (ROOT / "profiles" / "orchestrator" / "config.yaml").read_text()
    assert rendered.splitlines()[0] in cfg  # first rendered line is embedded in the route
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest scripts/test_render_deal_stage_route.py -v`
Expected: FAIL (renderer missing).

- [ ] **Step 3: Write the renderer**

Create `scripts/render-deal-stage-route.py`:

```python
"""render-deal-stage-route.py — render the deal-stage route prompt from deal_stages.yaml."""
import pathlib
import yaml

ROOT = pathlib.Path(__file__).resolve().parent.parent
MAP = ROOT / "workspace" / "shared" / "deal_stages.yaml"


def render() -> str:
    data = yaml.safe_load(MAP.read_text())
    lines = [
        "Deal HubSpot cambio de stage. Crear UNA task de kanban segun este mapa:",
    ]
    for stage, e in data["stages"].items():
        lines.append(f"  {stage} -> assignee=\"{e['domain']}\" task_type=\"{e['task_type']}\" ({e['name']})")
    lines += [
        "",
        "Deal id (usar como idempotency_key): {deal_id}",
        "Stage recibido: {dealstage}",
        "",
        "Llama kanban_create(assignee=<domain del mapa>, idempotency_key={deal_id},",
        "title=\"{dealname}\",",
        "metadata={\"deal_id\": \"{deal_id}\", \"dealstage\": \"{dealstage}\",",
        "\"domain\": <domain>, \"task_type\": <task_type>}).",
        "Si {dealstage} no esta en el mapa, NO crees task: responde con el error.",
    ]
    return "\n".join(lines)


if __name__ == "__main__":
    print(render())
```

- [ ] **Step 4: Run test to verify it fails on the config assertion**

Run: `python3 -m pytest scripts/test_render_deal_stage_route.py -v`
Expected: `test_renders_every_stage` and `test_mentions_no_terminal_tools` PASS; `test_config_matches_render` FAILS (route not yet in config).

- [ ] **Step 5: Add the route to config.yaml**

Render the prompt:

```bash
python3 scripts/render-deal-stage-route.py
```

In `profiles/orchestrator/config.yaml`, under `platforms.webhook.extra.routes:`, add (keep `onboarding-intake` as a thin alias that forwards to the same prompt):

```yaml
        deal-stage:
          profile: orchestrator
          secret: "${WEBHOOK_SECRET_DEAL_STAGE}"
          toolsets: [kanban]
          prompt: |
            # >>> generated by scripts/render-deal-stage-route.py — do not edit by hand
            Deal Hubspot cambio de stage. Crear UNA task de kanban segun este mapa:
              appointmentscheduled -> assignee="inbox" task_type="deal_qualified" (Calificar)
              qualifiedtobuy -> assignee="operations" task_type="deal_product_details_track" (Proyecto Viable)
              presentationscheduled -> assignee="orchestrator" task_type="deal_presentation_prep" (Presentacion de Solucion)
              decisionmakerboughtin -> assignee="inbox" task_type="deal_negotiation_monitor" (Negociacion/Revision)
              contractsent -> assignee="operations" task_type="deal_close_expectation" (Expectativa de Cierre)
              closedwon -> assignee="onboarding" task_type="onboarding_intake" (Ganado)
              closedlost -> assignee="operations" task_type="deal_lost_record" (Perdido)

            Deal id (usar como idempotency_key): {deal_id}
            Stage recibido: {dealstage}

            Llama kanban_create(assignee=<domain del mapa>, idempotency_key={deal_id},
            title="{dealname}",
            metadata={"deal_id": "{deal_id}", "dealstage": "{dealstage}",
            "domain": <domain>, "task_type": <task_type>}).
            Si {dealstage} no esta en el mapa, NO crees task: responde con el error.
            # <<< generated
          deliver: "log"
```

The block above is the exact expected renderer output; if `python3 scripts/render-deal-stage-route.py` differs, update both the renderer and this block so the test passes.

- [ ] **Step 6: Run test to verify it passes**

Run: `python3 -m pytest scripts/test_render_deal_stage_route.py -v`
Expected: 3 passed.

- [ ] **Step 7: Commit**

```bash
git add scripts/render-deal-stage-route.py scripts/test_render_deal_stage_route.py profiles/orchestrator/config.yaml
git commit -m "feat(pipeline): deal-stage orchestrator webhook route"
```

---

## Task 6: n8n signing branch and connected outputs

**Files:**

- Modify (live instance + repo snapshot): `n8n/factronics-forms-live/webhook-deal-stage.json`
- Test: `n8n/factronics-forms-live/test_webhook_deal_stage.py`

- [ ] **Step 1: Write the failing test**

Create `n8n/factronics-forms-live/test_webhook_deal_stage.py`:

```python
import json
import pathlib

HERE = pathlib.Path(__file__).parent
WF = json.loads((HERE / "webhook-deal-stage.json").read_text())

STAGES = ["Proyecto Viable", "Ganado", "Calificar", "Presentacion",
          "Negociacion/Revision", "Expectativa de Cierre", "Perdido"]


def _by_name(name):
    return next(n for n in WF["nodes"] if n["name"] == name)


def test_signing_node_exists():
    node = _by_name("Hermes - Firma HMAC V2 (deal-stage)")
    code = node["parameters"]["functionCode"]
    assert "createHmac" in code
    assert "dealstage" in code


def test_post_node_targets_deal_stage_route():
    node = _by_name("Hermes - POST deal-stage")
    url = node["parameters"]["url"]
    assert "/p/orchestrator/webhooks/deal-stage" in url


def test_every_switch_output_is_connected():
    switch = _by_name("Switch")
    out_names = [v["outputKey"] for v in switch["parameters"]["rules"]["values"]]
    conns = WF["connections"]["Switch"]["main"]
    assert len(conns) == len(out_names)
    for i, name in enumerate(out_names):
        assert i < len(conns) and conns[i], f"output {name} not connected"
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python3 -m pytest n8n/factronics-forms-live/test_webhook_deal_stage.py -v`
Expected: FAIL (nodes/connections missing).

- [ ] **Step 3: Apply the workflow changes**

Using n8n-mcp against the live workflow `oGMm7eDclq6VrVfW`:

1. Add node `Hermes - Firma HMAC V2 (deal-stage)` (`n8n-nodes-base.code`, `runOnceForAllItems`):

```javascript
const crypto = require('crypto');
const secret = $env.WEBHOOK_SECRET_DEAL_STAGE;
const timestamp = Math.floor(Date.now() / 1000).toString();
const out = [];
for (const item of $input.all()) {
  const raw = item.json.body || item.json;
  const events = Array.isArray(raw) ? raw : [raw];
  const stageEv = events.find(e => e && e.propertyName === 'dealstage');
  const dealId = (stageEv && (stageEv.objectId || stageEv.primaryObjectId))
    || (events[0] && events[0].objectId) || null;
  const dealstage = stageEv ? stageEv.propertyValue : null;
  const payload = { deal_id: dealId ? String(dealId) : null, dealstage, source: 'n8n:webhook-deal-stage' };
  const body = JSON.stringify(payload);
  const signature = crypto.createHmac('sha256', secret).update(`${timestamp}.${body}`).digest('hex');
  out.push({ json: { body, timestamp, signature } });
}
return out;
```

2. Add node `Hermes - POST deal-stage` (`n8n-nodes-base.httpRequest`, `onError: continueRegularOutput`):

```
method:  POST
url:     ={{$env.HERMES_WEBHOOK_BASE_URL}}/p/orchestrator/webhooks/deal-stage
headers: X-Webhook-Signature-V2 = ={{$json.signature}}
         X-Webhook-Timestamp   = ={{$json.timestamp}}
         Content-Type          = application/json
body:    json -> ={{$json.body}}
```

3. Connect **all seven** Switch outputs (`main[0..6]`) to `Hermes - Firma HMAC V2 (deal-stage)`, then that node to `Hermes - POST deal-stage`.
4. Set `WEBHOOK_SECRET_DEAL_STAGE` and `HERMES_WEBHOOK_BASE_URL` on the n8n host.
5. Validate and re-export:

```
n8n-mcp validate_workflow({ workflow: [[ORCA_RICH_MD:efc37ca260af7380159b7e078beab8d6:inline-html:%3Cjson%3E]] })   -> errorCount: 0
```

Then save the updated workflow as `n8n/factronics-forms-live/webhook-deal-stage.json`.

- [ ] **Step 4: Run test to verify it passes**

Run: `python3 -m pytest n8n/factronics-forms-live/test_webhook_deal_stage.py -v`
Expected: 3 passed.

- [ ] **Step 5: Commit**

```bash
git add n8n/factronics-forms-live/webhook-deal-stage.json n8n/factronics-forms-live/test_webhook_deal_stage.py
git commit -m "feat(pipeline): sign and route all seven stages to Hermes"
```

---

## Task 7: Docs and end-to-end smoke

**Files:**

- Modify: `AGENTS.md`
- Modify: `n8n/README.md`

- [ ] **Step 1: Update AGENTS.md routing section**

Add to the routing rules: a deal stage change arrives on `deal-stage`, and the orchestrator maps it via `workspace/shared/deal_stages.yaml`. Note `gateway` is never a routing target.

- [ ] **Step 2: Update n8n/README.md**

Move the `deal-stage` branch from "pending" to "defined and validated", and reference `scripts/render-deal-stage-route.py` as the source of the orchestrator prompt.

- [ ] **Step 3: End-to-end smoke test**

With `DATABASE_URL` set and the route deployed:

```bash
# 1. signed POST per stage (use a test deal id)
base="${HERMES_WEBHOOK_BASE_URL}/p/orchestrator/webhooks/deal-stage"
for s in appointmentscheduled qualifiedtobuy presentationscheduled decisionmakerboughtin contractsent closedwon closedlost; do
  body="{\"deal_id\":\"deal_test_001\",\"dealstage\":\"$s\",\"source\":\"smoke\"}"
  ts=$(date +%s)
  sig=$(printf '%s.%s' "$ts" "$body" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET_DEAL_STAGE" -hex | awk '{print $2}')
  curl -fsS -X POST "$base" \
    -H "Content-Type: application/json" \
    -H "X-Webhook-Signature-V2: $sig" \
    -H "X-Webhook-Timestamp: $ts" \
    --data "$body"
done

# 2. each stage produced a task and a deal_map row
for s in appointmentscheduled qualifiedtobuy presentationscheduled decisionmakerboughtin contractsent closedwon closedlost; do
  bash workspace/scripts/verify_deal_map.sh deal_test_001 "$s" || echo "MISSING $s"
done
```

Expected: `verify_deal_map.sh` exits 0 for all seven; a second delivery of the same stage creates no duplicate (idempotency).

- [ ] **Step 4: Commit**

```bash
git add AGENTS.md n8n/README.md
git commit -m "docs(pipeline): routing and bridge status"
```

---

## Definition of done

- All seven stage switches are connected and signed.
- The orchestrator creates a routed task per stage.
- `deal_map` has one row per (deal, stage), idempotent.
- `verify_deal_map.sh` passes for all seven stages on a test deal.
- No client-facing or QuickBooks write behavior changed.