# P5: New Form Processes 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:** Add three structured intakes (Close/PO Details, Quote/Negotiation Details, Loss Reason) by generalizing the existing Product Details and Equipment Installation form-to-document pattern.

**Architecture:** One registry maps a form to a Google Docs template, an output filename, and the deal attach mode. One reusable n8n sub-workflow does the mechanical work: read submission, merge into the template, export PDF, email the owner (Jovan copied), attach to the deal. Each new form is a registry entry plus a template.

**Tech Stack:** n8n, Google Docs API, Google Drive API, HubSpot, Python 3 (registry loader).

**Depends on:** P1. The Close/PO intake also feeds P (onboarding) via `deal_map` metadata.

**Spec:** `../specs/2026-09-24-sales-pipeline-hermes-design.md` (SP5, sections 5.5, 7).

---

## File Structure

| Path | Responsibility |
|---|---|
| `forms/registry.yaml` | Form to template and routing registry |
| `forms/load_registry.py` | Registry loader and validator |
| `forms/templates/*.docx.md` | Field-to-placeholder maps per form |
| `n8n/pipeline-form-to-document.json` | Reusable sub-workflow |
| `n8n/pipeline-close-po.json` | Close/PO Details wiring |
| `n8n/pipeline-quote-details.json` | Quote/Negotiation Details wiring |
| `n8n/pipeline-loss-reason.json` | Loss Reason wiring |

---

## Task 1: Form registry

**Files:**
- Create: `forms/registry.yaml`
- Create: `forms/load_registry.py`
- Test: `forms/test_load_registry.py`

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

Create `forms/test_load_registry.py`:

```python
import importlib.util
import pathlib

HERE = pathlib.Path(__file__).parent


def load():
    spec = importlib.util.spec_from_file_location("lr", HERE / "load_registry.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def test_expected_forms_present():
    mod = load()
    reg = mod.load()
    assert set(reg["forms"].keys()) == {
        "product_details", "equipment_installation",
        "close_po", "quote_details", "loss_reason",
    }


def test_each_form_has_template_and_output():
    mod = load()
    for name, f in mod.load()["forms"].items():
        assert f["template_doc_id"], name
        assert f["output_name"], name


def test_loss_reason_is_logged_not_emailed():
    mod = load()
    f = mod.load()["forms"]["loss_reason"]
    assert f["attach"] == "note"
    assert f["email_owner"] is False
```

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

Run: `python3 -m pytest forms/test_load_registry.py -v`
Expected: FAIL.

- [ ] **Step 3: Create the registry and loader**

Create `forms/registry.yaml`:

```yaml
# registry.yaml — structured intakes. template_doc_id is the Google Docs template.
version: 1
forms:
  product_details:
    stage: qualifiedtobuy
    template_doc_id: "<existing product details doc id>"
    output_name: "{{ dealname }} - Product Details.pdf"
    attach: file
    email_owner: true
  equipment_installation:
    stage: closedwon
    template_doc_id: "<existing equipment installation doc id>"
    output_name: "{{ dealname }} - Equipment Installation.pdf"
    attach: file
    email_owner: true
  close_po:
    stage: contractsent
    template_doc_id: "REPLACE_WITH_APPROVED_DOC_ID"
    output_name: "{{ dealname }} - Close and PO Details.pdf"
    attach: file
    email_owner: true
  quote_details:
    stage: decisionmakerboughtin
    template_doc_id: "REPLACE_WITH_APPROVED_DOC_ID"
    output_name: "{{ dealname }} - Quote Details.pdf"
    attach: file
    email_owner: true
  loss_reason:
    stage: closedlost
    template_doc_id: "REPLACE_WITH_APPROVED_DOC_ID"
    output_name: "{{ dealname }} - Loss Reason.pdf"
    attach: note
    email_owner: false
```

Create `forms/load_registry.py`:

```python
"""load_registry.py — read and validate the form registry."""
import pathlib
import yaml

HERE = pathlib.Path(__file__).parent


def load() -> dict:
    data = yaml.safe_load((HERE / "registry.yaml").read_text())
    for name, f in data["forms"].items():
        for key in ("stage", "template_doc_id", "output_name", "attach", "email_owner"):
            if key not in f:
                raise ValueError(f"{name} missing {key}")
        if f["attach"] not in ("file", "note"):
            raise ValueError(f"{name} bad attach mode")
    return data
```

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

Run: `python3 -m pytest forms/test_load_registry.py -v`
Expected: 3 passed. (The `REPLACE_WITH_APPROVED_DOC_ID` values are resolved in Tasks 3 to 5.)

- [ ] **Step 5: Commit**

```bash
git add forms/registry.yaml forms/load_registry.py forms/test_load_registry.py
git commit -m "feat(forms): registry and loader"
```

---

## Task 2: Reusable form-to-document sub-workflow

**Files:**
- Create: `n8n/pipeline-form-to-document.json`
- Test: `n8n/test_pipeline_form_to_document.py`

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

Create `n8n/test_pipeline_form_to_document.py`:

```python
import json
import pathlib

WF = json.loads((pathlib.Path(__file__).parent / "pipeline-form-to-document.json").read_text())
NAMES = [n["name"] for n in WF["nodes"]]


def test_steps_present():
    for step in ["Copy template", "Replace placeholders", "Export PDF", "Email owner", "Attach to deal"]:
        assert any(step in n for n in NAMES), step


def test_has_execute_workflow_trigger():
    assert any(n["type"] == "n8n-nodes-base.executeWorkflowTrigger" for n in WF["nodes"])
```

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

Run: `python3 -m pytest n8n/test_pipeline_form_to_document.py -v`
Expected: FAIL.

- [ ] **Step 3: Build the sub-workflow**

Create `n8n/pipeline-form-to-document.json` with an `executeWorkflowTrigger` and these steps:

1. **Code** `Normalize submission`: map form fields to a `fields` dict keyed by placeholder name.
2. **Google Drive** `Copy template`: `files.copy` the `template_doc_id` to a run folder, name it from `output_name`.
3. **Code** `Build replace requests`: turn `fields` into `replaceAllText` requests; error if a required placeholder is missing.
4. **Google Docs** `Replace placeholders`: `documents.batchUpdate`.
5. **Google Drive** `Export PDF`: export the doc to PDF into the deal folder.
6. **Gmail** `Email owner`: to owner, Jovan copied, PDF attached (skipped when `email_owner` is false).
7. **HubSpot** `Attach to deal`: upload the PDF as a file or create a note with the link, depending on `attach`.

Parameters come from the caller as `{ deal_id, dealname, owner_email, form, fields }`.

- [ ] **Step 4: Validate and run the test**

```
n8n-mcp validate_workflow({ workflow: <json> })   -> errorCount: 0
python3 -m pytest n8n/test_pipeline_form_to_document.py -v
```
Expected: 2 passed.

- [ ] **Step 5: Commit**

```bash
git add n8n/pipeline-form-to-document.json n8n/test_pipeline_form_to_document.py
git commit -m "feat(forms): reusable form-to-document sub-workflow"
```

---

## Task 3: Close/PO Details intake (feeds onboarding)

**Files:**
- Create: `forms/templates/close_po.md`
- Create: `n8n/pipeline-close-po.json`
- Test: `n8n/test_pipeline_close_po.py`

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

Create `n8n/test_pipeline_close_po.py`:

```python
import json
import pathlib

WF = json.loads((pathlib.Path(__file__).parent / "pipeline-close-po.json").read_text())
NAMES = [n["name"] for n in WF["nodes"]]


def test_triggers_subworkflow_and_marks_deal():
    assert any("Call form-to-document" in n for n in NAMES)
    assert any("Write close details" in n for n in NAMES)
```

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

Run: `python3 -m pytest n8n/test_pipeline_close_po.py -v`
Expected: FAIL.

- [ ] **Step 3: Create the field map and workflow**

Create `forms/templates/close_po.md` documenting the placeholders: `expected_close_date, po_number, payment_terms, delivery_date, billing_address, shipping_address, tax_exempt_status`.

Create `n8n/pipeline-close-po.json`: Form Trigger `Close/PO Details` (fields above), then `Call form-to-document` with `form: close_po`, then `Write close details` (HubSpot deal properties or a note) so `onboarding` can read them. Set the `template_doc_id` in `forms/registry.yaml` to the approved Google Doc ID.

- [ ] **Step 4: Validate and run the test**

```
n8n-mcp validate_workflow({ workflow: <json> })   -> errorCount: 0
python3 -m pytest n8n/test_pipeline_close_po.py -v
```
Expected: 1 passed.

- [ ] **Step 5: Commit**

```bash
git add forms/templates/close_po.md n8n/pipeline-close-po.json n8n/test_pipeline_close_po.py forms/registry.yaml
git commit -m "feat(forms): close/PO details intake"
```

---

## Task 4: Quote/Negotiation Details intake

**Files:**
- Create: `forms/templates/quote_details.md`
- Create: `n8n/pipeline-quote-details.json`
- Test: `n8n/test_pipeline_quote_details.py`

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

Create `n8n/test_pipeline_quote_details.py`:

```python
import json
import pathlib

WF = json.loads((pathlib.Path(__file__).parent / "pipeline-quote-details.json").read_text())
NAMES = [n["name"] for n in WF["nodes"]]


def test_has_form_and_subworkflow_and_optional_estimate():
    assert any("Quote" in n for n in NAMES)
    assert any("Call form-to-document" in n for n in NAMES)
```

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

Run: `python3 -m pytest n8n/test_pipeline_quote_details.py -v`
Expected: FAIL.

- [ ] **Step 3: Create the field map and workflow**

Create `forms/templates/quote_details.md` with placeholders: `scope, price, terms, leasing, competitors, decision_makers, objections`.

Create `n8n/pipeline-quote-details.json`: Form Trigger `Quote/Negotiation Details`, then `Call form-to-document` with `form: quote_details`, then an optional `Propose QB estimate` step that creates an approval item for `gateway` (reuse the `writeback_queue` approval pattern from P3). Set the approved `template_doc_id`.

- [ ] **Step 4: Validate and run the test**

```
n8n-mcp validate_workflow({ workflow: <json> })   -> errorCount: 0
python3 -m pytest n8n/test_pipeline_quote_details.py -v
```
Expected: 1 passed.

- [ ] **Step 5: Commit**

```bash
git add forms/templates/quote_details.md n8n/pipeline-quote-details.json n8n/test_pipeline_quote_details.py forms/registry.yaml
git commit -m "feat(forms): quote/negotiation details intake"
```

---

## Task 5: Loss Reason intake

**Files:**
- Create: `n8n/pipeline-loss-reason.json`
- Test: `n8n/test_pipeline_loss_reason.py`

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

Create `n8n/test_pipeline_loss_reason.py`:

```python
import json
import pathlib

WF = json.loads((pathlib.Path(__file__).parent / "pipeline-loss-reason.json").read_text())
NAMES = [n["name"] for n in WF["nodes"]]


def test_logs_and_does_not_email():
    assert any("Log loss reason" in n for n in NAMES)
    assert not any(n.startswith("Gmail") for n in WF["nodes"] if isinstance(n, str))
```

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

Run: `python3 -m pytest n8n/test_pipeline_loss_reason.py -v`
Expected: FAIL.

- [ ] **Step 3: Build the workflow**

Create `n8n/pipeline-loss-reason.json`: Form Trigger `Loss Reason` (fields: reason category one of price/timing/competitor/no_fit/no_response, free note), then `Log loss reason` to Postgres (`audit_log` metadata) and a HubSpot deal note. No email.

- [ ] **Step 4: Validate and run the test**

```
n8n-mcp validate_workflow({ workflow: <json> })   -> errorCount: 0
python3 -m pytest n8n/test_pipeline_loss_reason.py -v
```
Expected: 1 passed.

- [ ] **Step 5: Commit**

```bash
git add n8n/pipeline-loss-reason.json n8n/test_pipeline_loss_reason.py forms/registry.yaml
git commit -m "feat(forms): loss reason intake"
```

---

## Task 6: Docs and verification

- [ ] **Step 1:** Document resolved `template_doc_id` values and the field maps in `forms/README.md`.
- [ ] **Step 2:** End-to-end: submit each form, confirm the PDF is generated, emailed (except Loss Reason), and attached, and that Close/PO details are readable by `onboarding`.
- [ ] **Step 3:** Commit (`git commit -m "docs(forms): registry and template ids"`).

---

## Definition of done

- One registry drives all five intakes.
- The sub-workflow is reused by every form.
- Each form generates a branded PDF, emails the owner (except Loss Reason), and attaches to the deal.
- Close/PO details feed onboarding, removing double data entry.