kaiju_harness/business.py

#!/usr/bin/env python3
"""Business document harness for common solo-business deliverables.

The model can eventually plan these documents, but this renderer owns the
structure so invoices, proposals, launch plans, and email sequences are always
complete enough to hand to a customer.
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


@dataclass
class BusinessDocSpec:
    document_type: str
    business_name: str
    client_name: str = "Client"
    project_name: str = "Business project"
    amount: str = "$1,500"
    timeline: str = "1-2 weeks"
    deliverables: list[str] = field(default_factory=list)
    channels: list[str] = field(default_factory=list)
    tone: str = "clear, direct, professional"


DEFAULT_DELIVERABLES = [
    "Discovery and scope confirmation",
    "Complete first draft",
    "One revision pass",
    "Final handoff with next steps",
]

TECHNICAL_DOCUMENT_TYPES = {
    "implementation_plan",
    "debug_runbook",
    "automation_design",
    "tool_bridge_design",
    "computer_use_workflow",
    "release_checklist",
}


def clean_text(value: Any, fallback: str) -> str:
    if not isinstance(value, str):
        return fallback
    value = re.sub(r"\s+", " ", value).strip()
    return value or fallback


def infer_document_type(prompt: str) -> str:
    lower = prompt.lower()
    if any(term in lower for term in ["release checklist", "notarized", "dmg rebuild", "gatekeeper"]):
        return "release_checklist"
    if any(term in lower for term in ["computer-use workflow", "upload a youtube", "screenshots/checkpoints", "final video url"]):
        return "computer_use_workflow"
    if any(term in lower for term in [
        "telegram coding-agent bridge",
        "telegram bridge",
        "state model",
        "duplicate messages",
        "diff review ui",
        "file diff review",
        "per-hunk",
        "approve/reject/apply/revert",
        "changed-file summary",
    ]):
        return "tool_bridge_design"
    if any(term in lower for term in [
        "jah premium credits",
        "safe web-search proxy",
        "perplexity",
        "rate-limits per customer",
        "rate-limit abuse",
        "license gate",
        "local model fleet router",
        "fleet router",
        "update-check service",
        "safe update service",
        "phased rollout",
        "webhook flow",
        "reserve credits",
        "debit actual",
        "worker pseudo-code",
        "pseudocode",
    ]):
        return "automation_design"
    if any(term in lower for term in ["diagnose", "root cause", "times out", "opening the file fails", "patch strategy", "verification checklist"]):
        return "debug_runbook"
    if any(term in lower for term in ["cli update command", "provide file structure", "shell code", "test plan", "artifact-writing layer", "artifact writing layer", "write files safely", "quoted shell commands"]):
        return "implementation_plan"
    if any(term in lower for term in ["invoice", "bill", "payment request"]):
        return "invoice"
    if any(term in lower for term in ["proposal", "scope", "quote"]):
        return "proposal"
    if any(term in lower for term in ["launch plan", "go-to-market", "marketing plan", "promotion plan"]):
        return "launch_plan"
    if any(term in lower for term in ["email sequence", "follow-up", "welcome email", "sales email"]):
        return "email_sequence"
    return "business_brief"


def infer_name_after(prompt: str, markers: list[str], fallback: str) -> str:
    stop_words = r"\s+(?:for|with|to|that|using|include|including|trying|attempting|looking|who|from)\b"
    for marker in markers:
        pattern = rf"{marker}\s+([A-Z][A-Za-z0-9 &'&-]{{2,70}}?)(?:\.|,|{stop_words}|$)"
        match = re.search(pattern, prompt, flags=re.IGNORECASE)
        if match:
            return clean_text(match.group(1), fallback).rstrip(".")
    return fallback


def infer_amount(prompt: str) -> str:
    match = re.search(r"\$\s?[0-9][0-9,]*(?:\.\d{2})?", prompt)
    if match:
        return match.group(0).replace("$ ", "$")
    return "$1,500"


def infer_deliverables(prompt: str, document_type: str) -> list[str]:
    lower = prompt.lower()
    if "website" in lower and "stripe" in lower:
        return ["One-page responsive website", "Conversion-focused copy", "Stripe checkout setup", "Webhook/payment test plan"]
    if "stripe" in lower:
        return ["Stripe checkout setup", "Webhook handling", "Payment confirmation", "Basic test plan"]
    if "website" in lower or "landing page" in lower:
        return ["One-page responsive website", "Conversion-focused copy", "Contact or booking section", "Launch checklist"]
    if "roof" in lower or "storm" in lower:
        return ["Confirm storm inspection need", "Show trust proof", "Offer inspection window", "Close with direct booking CTA"]
    if "social" in lower or "youtube" in lower:
        return ["Launch video angle", "Posting calendar", "Short-form clips", "Follow-up offer"]
    if document_type == "email_sequence":
        return ["Email 1: first touch", "Email 2: value proof", "Email 3: direct call to action"]
    return DEFAULT_DELIVERABLES.copy()


def infer_channels(prompt: str) -> list[str]:
    lower = prompt.lower()
    channels = []
    for channel in ["YouTube", "LinkedIn", "Email", "Discord", "Google Business", "Instagram"]:
        if channel.lower() in lower:
            channels.append(channel)
    return channels or ["YouTube", "Email", "Google Business"]


def spec_from_prompt(prompt: str) -> BusinessDocSpec:
    document_type = infer_document_type(prompt)
    business_name = infer_name_after(prompt, ["named", "for", "business"], "Kaiju" if document_type in TECHNICAL_DOCUMENT_TYPES else "Local Business")
    client_name = infer_name_after(prompt, ["client named", "customer named"], "Client")
    project_name = "Website build" if "website" in prompt.lower() else "Business project"
    if document_type == "implementation_plan":
        project_name = "CLI update command"
    if document_type == "debug_runbook":
        project_name = "Production debugging runbook"
    if document_type == "automation_design":
        project_name = "Backend automation design"
    if document_type == "tool_bridge_design":
        project_name = "Telegram bridge design"
    if document_type == "computer_use_workflow":
        project_name = "Computer-use workflow"
    if document_type == "release_checklist":
        project_name = "Mac release checklist"
    if "launch" in prompt.lower():
        project_name = "Launch campaign"
    if "roof" in prompt.lower() or "storm" in prompt.lower():
        project_name = "Storm inspection follow-up"
    return BusinessDocSpec(
        document_type=document_type,
        business_name=business_name,
        client_name=client_name,
        project_name=project_name,
        amount=infer_amount(prompt),
        deliverables=infer_deliverables(prompt, document_type),
        channels=infer_channels(prompt),
    )


def normalize_spec(raw: dict[str, Any] | BusinessDocSpec, prompt: str = "") -> BusinessDocSpec:
    inferred_type = infer_document_type(prompt)
    if isinstance(raw, BusinessDocSpec):
        spec = raw
    else:
        fallback = spec_from_prompt(prompt)
        deliverables = raw.get("deliverables") if isinstance(raw.get("deliverables"), list) else fallback.deliverables
        channels = raw.get("channels") if isinstance(raw.get("channels"), list) else fallback.channels
        spec = BusinessDocSpec(
            document_type=clean_text(raw.get("document_type"), fallback.document_type).lower(),
            business_name=clean_text(raw.get("business_name"), fallback.business_name),
            client_name=clean_text(raw.get("client_name"), fallback.client_name),
            project_name=clean_text(raw.get("project_name"), fallback.project_name),
            amount=clean_text(raw.get("amount"), fallback.amount),
            timeline=clean_text(raw.get("timeline"), fallback.timeline),
            deliverables=[clean_text(item, "") for item in deliverables if isinstance(item, str)][:8] or fallback.deliverables,
            channels=[clean_text(item, "") for item in channels if isinstance(item, str)][:6] or fallback.channels,
            tone=clean_text(raw.get("tone"), fallback.tone),
        )
    allowed_types = {"invoice", "proposal", "launch_plan", "email_sequence", "business_brief"} | TECHNICAL_DOCUMENT_TYPES
    if spec.document_type not in allowed_types:
        spec.document_type = inferred_type
    if inferred_type in TECHNICAL_DOCUMENT_TYPES and spec.document_type not in TECHNICAL_DOCUMENT_TYPES:
        # The planner can collapse technical architecture/debug prompts into a
        # generic business brief. Keep the model's useful names if present, but
        # never let a shallow document type bypass the deterministic templates.
        spec.document_type = inferred_type
        fallback = spec_from_prompt(prompt)
        if spec.business_name in {"Local Business", "Business", "Kaiju"}:
            spec.business_name = fallback.business_name
        spec.project_name = fallback.project_name
    if not spec.deliverables:
        spec.deliverables = DEFAULT_DELIVERABLES.copy()
    if not spec.channels:
        spec.channels = ["YouTube", "Email", "Google Business"]
    return spec


def bullet_list(items: list[str]) -> str:
    return "\n".join(f"- {item}" for item in items)


def render_invoice(spec: BusinessDocSpec) -> str:
    rows = "\n".join(f"| {item} | Included |" for item in spec.deliverables)
    return f"""# Invoice - {spec.business_name}

**Client:** {spec.client_name}  
**Project:** {spec.project_name}  
**Due:** Upon receipt  
**Total:** {spec.amount}

## Services

| Item | Status |
| --- | --- |
{rows}

## Payment

Pay by card, ACH, or the secure payment link provided by {spec.business_name}. Payment is required before final handoff unless another agreement is signed.

## Notes

- Scope changes are quoted before work continues.
- Final files, access, or launch assets are released after payment clears.
- Questions should be sent in one thread so the project stays organized.
"""


def render_proposal(spec: BusinessDocSpec) -> str:
    return f"""# Proposal - {spec.project_name}

**Prepared for:** {spec.client_name}  
**Prepared by:** {spec.business_name}  
**Timeline:** {spec.timeline}  
**Investment:** {spec.amount}

## Outcome

This project creates a clean, useful result that helps the business get customers to the next step without confusion.

## Scope

{bullet_list(spec.deliverables)}

## Process

1. Confirm the scope, assets, and success criteria.
2. Build the first complete version.
3. Review once against the business goal.
4. Launch or hand off with clear next steps.

## What Success Looks Like

- The customer understands the offer quickly.
- The owner knows what was delivered and how to use it.
- The next action is obvious.
- The project does not expand without approval.

## Approval

Reply with approval and the preferred start date. Once approved, {spec.business_name} will send payment details and the project checklist.
"""


def render_launch_plan(spec: BusinessDocSpec) -> str:
    channels = "\n".join(f"- **{channel}:** one clear post that shows the problem, the result, and the next step." for channel in spec.channels)
    return f"""# Launch Plan - {spec.business_name}

**Offer:** {spec.project_name}  
**Timeline:** {spec.timeline}  
**Goal:** get qualified people to understand the offer and take one action.

## Positioning

Lead with the customer pain, show the transformation, then make the offer simple. Avoid feature dumping.

## Channels

{channels}

## 7-Day Schedule

| Day | Action |
| --- | --- |
| 1 | Publish the main explanation video or post. |
| 2 | Share a customer-style example or demo. |
| 3 | Answer the biggest objection directly. |
| 4 | Post proof, screenshots, or before/after. |
| 5 | Send a concise email to warm contacts. |
| 6 | Repost the clearest demo with a stronger hook. |
| 7 | Review clicks, replies, sales, and support questions. |

## Metrics

- Sales or booked calls.
- Reply rate.
- Demo views and retention.
- Most common objection.
- Support friction after purchase.
"""


def render_email_sequence(spec: BusinessDocSpec) -> str:
    return f"""# Email Sequence - {spec.business_name}

## Email 1 - Simple First Touch

**Subject:** Quick idea for {spec.client_name}

Hi {spec.client_name},

I noticed an opportunity to make the next step easier for your customers. {spec.business_name} can help with {spec.project_name} in a practical way without turning it into a drawn-out project.

Would you like me to send the simple version of the plan?

## Email 2 - Value Proof

**Subject:** What this would actually change

The point is not more software or more meetings. The point is a cleaner path from interest to action.

What we would handle:

{bullet_list(spec.deliverables)}

If this is useful, I can map the first version and price it clearly.

## Email 3 - Direct Close

**Subject:** Should I close this out?

I do not want to keep bothering you. If {spec.project_name} still matters, reply with "send it" and I will send the clean next step.

If not, no problem. I will close the loop.
"""


def render_business_brief(spec: BusinessDocSpec) -> str:
    return f"""# Business Brief - {spec.business_name}

## Goal

Move {spec.project_name} forward with a simple plan, clear deliverables, and a next action.

## Scope

This brief is for a small business owner who needs a practical offer, not a vague strategy document. The scope should stay narrow enough to sell, fulfill, support, and explain in one customer conversation.

## Deliverables

{bullet_list(spec.deliverables)}

## Price And Payment Logic

- Use one clear starting price or package so the buyer understands the offer quickly.
- Require payment or deposit before implementation work begins.
- Keep support boundaries explicit so the service does not become unlimited unpaid consulting.
- Position the value around time saved, fewer missed leads, and a cleaner operating workflow.

## Recommendation

Start with the smallest version that can be sold, tested, or shown to a real customer. Do not add more tools until the offer is clear.

## Call To Action

Ask the customer to reply with the one workflow they want fixed first, then book the setup call or pay the deposit.

## Next Step

Pick one owner, one deadline, and one customer-facing outcome.
"""


def render_implementation_plan(spec: BusinessDocSpec, prompt: str) -> str:
    lower = prompt.lower()
    if "artifact" in lower or "write files safely" in lower:
        return f"""# Technical Implementation Plan - Artifact Writing Layer

## Goal

Build the artifact-writing layer for a desktop AI app so generated files are written safely, verified before the app claims success, and opened with customer-friendly errors when macOS permissions or paths fail.

## File Structure

```text
src/artifacts/
  artifact-paths.ts
  artifact-writer.ts
  open-file.ts
  permissions.ts
tests/
  artifact-writer.test.ts
  open-file.test.ts
```

## Shell Code And TypeScript Pattern

```ts
import {{ mkdir, stat, writeFile }} from "node:fs/promises";
import {{ dirname, resolve }} from "node:path";
import {{ spawn }} from "node:child_process";

export async function writeVerifiedArtifact(path: string, contents: string) {{
  const target = resolve(path);
  await mkdir(dirname(target), {{ recursive: true }});
  await writeFile(target, contents, "utf8");
  const info = await stat(target);
  if (!info.isFile() || info.size === 0) throw new Error(`Artifact was not written: ${{target}}`);
  return {{ path: target, size: info.size }};
}}

export async function openArtifact(path: string) {{
  const target = resolve(path);
  await stat(target);
  return new Promise<void>((resolveOpen, rejectOpen) => {{
    const child = spawn("open", ["--", target], {{ stdio: "ignore" }});
    child.on("exit", code => code === 0 ? resolveOpen() : rejectOpen(new Error(`open failed: ${{code}}`)));
    child.on("error", rejectOpen);
  }});
}}
```

## State And Config

- Store the artifact root in app settings, not in model text.
- Keep a manifest with prompt ID, file path, byte size, checksum, and open status.
- Preserve customer-selected folders across sessions.
- Record permission failures separately from generation failures.

## Safety Rules

- Use atomic writes: write to a temporary file, verify it, then rename into place for important artifacts.
- Never concatenate shell strings.
- Always pass paths as argv, for example `open -- "$path"`.
- Verify with `stat` before showing success.
- If Desktop/Documents access fails, write to the app artifact directory and show the exact path.
- Never delete or overwrite existing customer files without explicit confirmation.

## Tests

1. Write an HTML file and verify byte size.
2. Try a path with spaces and confirm `open --` works.
3. Simulate denied Desktop permission and confirm fallback directory is used.
4. Simulate empty write and confirm the app reports failure.
5. Confirm the manifest records path, checksum, and open status.

## Next Step

Implement the writer first, then wire it to the UI after the failure-path tests pass.
"""
    return f"""# Technical Implementation Plan - {spec.project_name}

## Goal

Build the Mac-only CLI update command as a safe customer-facing operation. It must reuse the saved license, verify the downloaded artifact, install atomically, preserve config, and print clear status without exposing secrets.

## File Structure

```text
bin/
  kiyomi-update
lib/
  license.sh
  download.sh
  checksum.sh
  install.sh
tests/
  update-smoke.bats
```

## Shell Code

```bash
#!/usr/bin/env bash
set -euo pipefail

APP_HOME="${{KIYOMI_HOME:-$HOME/.kiyomi}}"
LICENSE_FILE="$APP_HOME/license.key"
CONFIG_DIR="$APP_HOME/config"
WORK_DIR="$(mktemp -d)"
trap 'rm -rf "$WORK_DIR"' EXIT

status() {{ printf 'Kiyomi update: %s\\n' "$1"; }}

if [[ ! -s "$LICENSE_FILE" ]]; then
  status "No saved license key found. Run the installer first."
  exit 2
fi

LICENSE_KEY="$(tr -d '\\n' < "$LICENSE_FILE")"
status "Checking latest version..."

curl -fsSL \
  -H "x-license-key: $LICENSE_KEY" \
  -o "$WORK_DIR/latest.zip" \
  "https://updates.example.com/kiyomi/latest.zip"

curl -fsSL -o "$WORK_DIR/latest.sha256" \
  "https://updates.example.com/kiyomi/latest.zip.sha256"

(cd "$WORK_DIR" && shasum -a 256 -c latest.sha256)

status "Installing atomically..."
unzip -q "$WORK_DIR/latest.zip" -d "$WORK_DIR/new"
test -x "$WORK_DIR/new/bin/kiyomi"

mkdir -p "$APP_HOME/releases"
NEW_RELEASE="$APP_HOME/releases/$(date +%Y%m%d%H%M%S)"
mv "$WORK_DIR/new" "$NEW_RELEASE"
ln -sfn "$NEW_RELEASE" "$APP_HOME/current.tmp"
mv -Tf "$APP_HOME/current.tmp" "$APP_HOME/current"
mkdir -p "$CONFIG_DIR"

status "Update installed. Config preserved at $CONFIG_DIR."
```

## Safety Rules

- Never print the license key.
- Verify checksum before installing.
- Install into a new release directory, then switch the symlink.
- Preserve `config/`, logs, and customer data.
- Keep the previous release available for rollback.

## Tests

1. Run with no license file and verify the command exits with a clear message.
2. Run with a fake checksum and verify install is blocked.
3. Run with a valid zip and verify `current` points to the new release.
4. Verify config files survive the update.
5. Roll back by repointing `current` to the previous release.

## Next Step

Add the command behind one manual smoke test before shipping it to customers.
"""


def render_debug_runbook(spec: BusinessDocSpec, prompt: str) -> str:
    lower = prompt.lower()
    file_open = "opening the file fails" in lower or "desktop/index.html" in lower
    if file_open:
        causes = [
            "Path mismatch: the app reports `/Users/customer/Desktop/index.html`, but the write happened in a sandbox or different working directory.",
            "Shell quoting bug: spaces or special characters in the path were not quoted correctly.",
            "macOS permission issue: Desktop/Documents access was denied to the process that wrote or opened the file.",
            "Artifact write finished after the open command was attempted.",
            "The app trusted the model text instead of verifying the file exists on disk.",
        ]
        patch = [
            "Resolve all customer paths through a single path utility and reject ambiguous relative paths.",
            "Write the file, call `stat`, and only then show the open action.",
            "Use `open -- \"$path\"` and never concatenate shell strings.",
            "If Desktop access fails, write to the app artifact directory and show that exact path.",
            "Add a visible error with the attempted path, actual path, and permission hint.",
        ]
    else:
        causes = [
            "Model routing may be hitting a slow or wrong backend instead of the intended local model.",
            "Proxy timeout may be shorter than the artifact timeout required for website builds.",
            "Stream handling may hide progress until the end, making a healthy run look stalled.",
            "Token or output caps may stop the run before complete HTML is written.",
            "Artifact writing may happen after generation without verification or recovery.",
        ]
        patch = [
            "Log the selected provider/model at request start and final response.",
            "Separate normal chat timeout from artifact/build timeout.",
            "Stream status every 15-30 seconds while the task is active.",
            "Raise artifact output budget only for build-class tasks.",
            "Verify the artifact exists, is complete, and can be opened before claiming success.",
        ]
    return f"""# Debug Runbook - {spec.project_name}

## Likely Root Causes

{bullet_list(causes)}

## Safe Patch Plan

{bullet_list(patch)}

## Verification Checklist

- Reproduce the issue with the original customer prompt.
- Confirm routing shows the expected model/provider before the run starts.
- Confirm progress streams while work is active.
- Confirm the artifact path is absolute and exists with `stat`.
- Confirm timeout values match the task class.
- Confirm rollback path is documented before deploy.

## Rollback

If the patch increases failures, revert only the routing/timeout/artifact change set. Do not change customer licenses, billing, or unrelated proxy bindings during rollback.

## Next Step

Patch one layer at a time, run a website build, a file-open test, and a failure-path test before releasing.
"""


def render_automation_design(spec: BusinessDocSpec, prompt: str) -> str:
    lower = prompt.lower()
    if "search" in lower or "perplexity" in lower:
        title = "Safe Search Proxy"
        schema = """```sql
CREATE TABLE search_usage (
  id TEXT PRIMARY KEY,
  license_id TEXT NOT NULL,
  query TEXT NOT NULL,
  result_count INTEGER NOT NULL,
  status TEXT NOT NULL,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
```"""
        lifecycle = [
            "Client sends license header and search query to the Worker.",
            "Worker validates license status in D1 before calling any provider.",
            "Worker rate-limits per customer using KV or D1 counters.",
            "Worker calls Perplexity server-side with the hidden provider API key.",
            "Worker normalizes titles, URLs, snippets, and citations.",
            "Worker logs usage and falls back to a safe degraded provider if Perplexity fails.",
        ]
        pseudo = """```ts
export default {
  async fetch(request, env, ctx) {
    const license = await validateLicense(request, env.DB);
    await enforceRateLimit(license.id, env.SEARCH_LIMITS);
    const results = await callPerplexity(env.PERPLEXITY_API_KEY, await request.json());
    ctx.waitUntil(logUsage(env.DB, license.id, results));
    return Response.json({ results: normalize(results) });
  }
}
```"""
        safety = [
            "Never ship the Perplexity key in the client.",
            "Deploy with full Worker bindings intact.",
            "Cut off abusive customers by license ID.",
            "Keep a rollback Worker version ready.",
        ]
    elif "fleet" in lower or "goku" in lower or "fallback" in lower and "cloud" in lower:
        title = "Local Model Fleet Router"
        schema = """```sql
CREATE TABLE model_nodes (
  id TEXT PRIMARY KEY,
  public_name TEXT NOT NULL,
  private_route TEXT NOT NULL,
  status TEXT NOT NULL,
  priority INTEGER NOT NULL,
  last_health_at TEXT,
  latency_ms INTEGER,
  failure_count INTEGER DEFAULT 0
);

CREATE TABLE model_requests (
  id TEXT PRIMARY KEY,
  license_id TEXT NOT NULL,
  task_class TEXT NOT NULL,
  selected_node_id TEXT,
  status TEXT NOT NULL,
  started_at TEXT DEFAULT CURRENT_TIMESTAMP,
  finished_at TEXT
);
```"""
        lifecycle = [
            "Client sends a licensed request with a task class, not a machine name.",
            "Router checks node health and selects the first healthy local backend in priority order.",
            "Request streams customer-visible progress while the backend works.",
            "Timeout or transport failure marks the attempt failed and tries the next healthy backend.",
            "Loop prevention stops after one attempt per backend and then uses the cloud fallback policy.",
            "Usage, latency, selected node, failure reason, and final status are logged server-side.",
        ]
        pseudo = """```ts
async function routeModelRequest(request, env) {
  const license = await validateLicense(request, env.DB);
  const nodes = await healthyNodes(env.DB, ["standard", "fallback", "cloud"]);
  for (const node of nodes) {
    const attempt = await startAttempt(license.id, node.id, env.DB);
    try {
      return await streamFromNode(node.private_route, request, attempt);
    } catch (error) {
      await markFailed(attempt, error, env.DB);
    }
  }
  return new Response("No healthy model route available", { status: 503 });
}
```"""
        safety = [
            "Keep private machine names and provider API keys hidden from customers.",
            "Expose customer-friendly labels like Arianna Standard and Arianna Recovery.",
            "Use health checks before routing, not after the customer waits.",
            "Never loop forever; cap attempts and return a clear recoverable error.",
        ]
    elif "license gate" in lower or "license" in lower and "rate" in lower:
        title = "Cloudflare License Gate"
        schema = """```sql
CREATE TABLE licenses (
  id TEXT PRIMARY KEY,
  key_hash TEXT NOT NULL UNIQUE,
  customer_email TEXT,
  status TEXT NOT NULL,
  product TEXT NOT NULL,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP,
  revoked_at TEXT
);

CREATE TABLE license_usage (
  id TEXT PRIMARY KEY,
  license_id TEXT NOT NULL,
  route TEXT NOT NULL,
  status TEXT NOT NULL,
  latency_ms INTEGER,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
```"""
        lifecycle = [
            "Worker receives request with license header.",
            "Worker hashes the submitted key and validates active status in D1.",
            "Worker applies per-license rate limit before calling any private origin.",
            "Worker forwards only validated requests to the model/search/update service.",
            "Worker logs usage and returns clear errors for revoked, missing, or rate-limited keys.",
            "Deploy process verifies all D1/KV/R2 bindings before replacing production.",
        ]
        pseudo = """```ts
export default {
  async fetch(request, env, ctx) {
    const license = await validateLicenseHeader(request, env.DB);
    await enforceRateLimit(license.id, env.RATE_LIMITS);
    const response = await fetchPrivateOrigin(request, env.ORIGIN_SECRET);
    ctx.waitUntil(logUsage(env.DB, license.id, request.url, response.status));
    return response;
  }
}
```"""
        safety = [
            "Never store raw license keys, only hashes.",
            "Keep origin secrets and provider API keys server-side and hidden.",
            "Fail closed if bindings are missing.",
            "Keep rollback Worker version ready before deploy.",
        ]
    elif "update" in lower or "version" in lower or "checksum" in lower:
        title = "Safe Update Check Service"
        schema = """```json
{
  "version": "0.3.0",
  "downloadUrl": "https://downloads.example.com/Kiyomi-0.3.0.dmg",
  "sha256": "expected_checksum_here",
  "releaseNotes": ["Fixes artifact opening", "Improves streaming status"],
  "phasedRolloutPercent": 25,
  "minimumSupportedVersion": "0.2.0",
  "rollbackVersion": "0.2.9"
}
```"""
        lifecycle = [
            "App checks update endpoint with app version, product, license hash, and platform.",
            "Worker validates license/auth before returning private download metadata.",
            "Worker returns version, checksum, release notes, rollout flag, and rollback version.",
            "Client downloads the artifact and verifies checksum before presenting install.",
            "If phased rollout excludes the user, return current version with no update.",
            "If the update is pulled, return rollback metadata and a customer-safe message.",
        ]
        pseudo = """```ts
async function handleUpdateCheck(request, env) {
  const license = await validateLicenseHeader(request, env.DB);
  const current = await request.json();
  const release = await loadRelease(env.KV, current.product);
  if (!isInRollout(license.id, release.phasedRolloutPercent)) {
    return Response.json({ updateAvailable: false });
  }
  return Response.json({ updateAvailable: true, ...publicReleaseFields(release) });
}
```"""
        safety = [
            "Keep signing keys, origin secrets, and provider API keys server-side and hidden.",
            "Never force a blind update without checksum verification.",
            "Support rollback by version.",
            "Return clear errors when metadata is missing instead of sending a broken download.",
        ]
    else:
        title = "Jah Premium Credits"
        schema = """```sql
CREATE TABLE jah_accounts (
  id TEXT PRIMARY KEY,
  license_id TEXT NOT NULL UNIQUE,
  balance_cents INTEGER NOT NULL DEFAULT 0,
  status TEXT NOT NULL DEFAULT 'active'
);

CREATE TABLE jah_usage (
  id TEXT PRIMARY KEY,
  account_id TEXT NOT NULL,
  reserve_cents INTEGER NOT NULL,
  debit_cents INTEGER DEFAULT 0,
  refund_cents INTEGER DEFAULT 0,
  provider_cost_cents INTEGER DEFAULT 0,
  margin_multiplier REAL NOT NULL DEFAULT 3.0,
  status TEXT NOT NULL,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
```"""
        lifecycle = [
            "Stripe Checkout creates a top-up session for the selected credit amount.",
            "Webhook verifies the Stripe signature before crediting D1.",
            "Premium request reserves estimated credits before model call.",
            "Provider cost is measured, multiplied by 3x margin, and debited.",
            "Unused reserve is refunded immediately.",
            "Zero or negative balance is rejected before inference starts.",
        ]
        pseudo = """```ts
async function handlePremium(request, env) {
  const account = await loadAccount(request.headers.get("x-license-key"), env.DB);
  const reserve = await reserveCredits(account, 500, env.DB);
  try {
    const result = await callPremiumModel(request, env.PREMIUM_PROVIDER_KEY);
    await debitActualCost(reserve, result.usage.providerCostCents * 3, env.DB);
    return result.response;
  } catch (error) {
    await refundReserve(reserve, env.DB);
    throw error;
  }
}
```"""
        safety = [
            "Do not trust client payment status.",
            "Never allow negative balances.",
            "Never expose OpenAI, Anthropic, Gemini, or other provider API keys; keep provider secrets server-side and hidden from clients.",
            "Refund failed provider calls and log the request ID.",
        ]
    return f"""# Automation Design - {title}

## Data Model

{schema}

## Request Lifecycle

{bullet_list(lifecycle)}

## Pseudo-Code

{pseudo}

## Abuse Controls

- Validate auth or license before provider calls.
- Rate limit per customer.
- Log request ID, account ID, model/provider, status, reserve, debit, and refund.
- Cut off revoked or exhausted accounts before work starts.

## Failure Handling

{bullet_list(safety)}

## Deployment Notes

- Deploy to staging first.
- Verify D1/KV/R2 bindings before production deploy.
- Keep rollback command and previous Worker version ready.

## Next Step

Build the Worker endpoint behind a staging route, run one successful request, one exhausted-account request, and one provider-failure refund test before production.
"""


def render_tool_bridge_design(spec: BusinessDocSpec) -> str:
    return f"""# Tool Bridge Design - {spec.project_name}

## Goal

Make long Telegram coding-agent tasks usable by acknowledging immediately, showing progress, recovering after restart, and ending with a clear final summary.

## State Model

```text
task(id, chat_id, user_id, prompt, status, current_step, last_message_id, dedupe_key, created_at, updated_at)
task_event(id, task_id, kind, message, file_path, command, created_at)
```

## Flow

1. Receive Telegram message and create an idempotent task from `chat_id + message_id`.
2. Send immediate acknowledgement: "I started this and will update you while I work."
3. Run the agent loop and append tool steps, commands, and file changes to `task_event`.
4. Every 20-30 seconds, send a concise progress message if the visible state changed.
5. On restart, load active tasks and resume from the last durable step.
6. Send final summary with changed files, tests run, artifact links, and next step.

## Pseudo-Code

```ts
if (await seen(message.id)) return;
const task = await createTask(message);
await telegram.send(task.chatId, "Started. I will post progress every 20-30 seconds.");
for await (const event of runAgent(task.prompt)) {{
  await saveEvent(task.id, event);
  if (shouldSendProgress(task)) await sendProgress(task);
}}
await telegram.send(task.chatId, finalSummary(task));
```

## Duplicate Prevention

- Store Telegram `message_id` and a dedupe key.
- Make progress updates idempotent by step number.
- Never resend the same final summary after restart.

## Next Step

Build the bridge with durable state first; streaming every token is unnecessary and noisy.
"""


def render_computer_use_workflow(spec: BusinessDocSpec) -> str:
    return f"""# Computer-Use Workflow - YouTube Upload

## Goal

Upload a YouTube video from a local file with thumbnail, title, description, tags, visibility, and scheduled publish time while showing visible status and verifying the final URL.

## Browser Workflow

1. Confirm local video and thumbnail paths exist before opening the browser.
2. Open YouTube Studio and checkpoint the current screen.
3. If login is required, pause with a clear status instead of pretending work continued.
4. Click Create, upload the video, and wait for processing indicators.
5. Fill title, description, tags, thumbnail, playlist, visibility, and schedule.
6. Screenshot/checkpoint before final publish or schedule confirmation.
7. Confirm the final video URL from YouTube Studio and return it.

## Status Updates

- "Opening YouTube Studio."
- "Upload started; waiting for processing."
- "Metadata filled; verifying thumbnail."
- "Scheduling/publishing now."
- "Verified final URL."

## Failure Recovery

- If upload stalls, retry once and preserve the current draft.
- If login expires, stop and ask for login action.
- If processing fails, report the exact YouTube status.
- Do not claim success without a final URL.

## Verification

The final answer must include the YouTube URL, scheduled time or visibility, and any warnings from YouTube processing.

## Next Step

Run the workflow once in a test browser profile and keep screenshots for each checkpoint before using it on a real customer upload.
"""


def render_release_checklist(spec: BusinessDocSpec) -> str:
    return f"""# Release Checklist - Notarized Mac DMG

## First Decision

If the change is only server-side auth proxy behavior, a DMG rebuild is not required. If the app binary, bundled runtime, entitlements, permissions, updater, or install flow changed, rebuild and notarize the DMG.

## Likely Root Causes To Diagnose

- Server fix only: proxy route, model routing, timeout, billing, or search behavior changed.
- App fix: local UI, file opening, permissions, bundled scripts, icon, updater, or entitlements changed.
- Packaging fix: signing identity, hardened runtime, notarization, staple, or Gatekeeper check failed.

## Smoke Tests

- Fresh install on a clean Mac user account.
- License entry and first request.
- Website build with artifact open.
- Computer-use status card if enabled.
- Premium/credits button opens the correct Stripe flow.

## Signing And Notarization

```bash
codesign --verify --deep --strict Kiyomi.app
spctl --assess --type execute --verbose Kiyomi.app
xcrun stapler validate Kiyomi.app
```

## Rollback

- Keep previous DMG and Worker version.
- If server-only fix fails, roll back the Worker, not the DMG.
- If binary fix fails, pull the download link and restore the previous DMG.

## Customer Install Test

Download from the real customer link, drag into Applications, launch normally, and verify Gatekeeper does not block it.

## Support Notes

Tell customers whether they need to download a new app or whether the server fix is already live. Do not force a reinstall for a proxy-only fix.

## Next Step

Run the customer install test from the actual public download link before announcing the release.
"""


def render_markdown(raw_spec: dict[str, Any] | BusinessDocSpec, prompt: str = "") -> str:
    spec = normalize_spec(raw_spec, prompt)
    if spec.document_type == "implementation_plan":
        return render_implementation_plan(spec, prompt)
    if spec.document_type == "debug_runbook":
        return render_debug_runbook(spec, prompt)
    if spec.document_type == "automation_design":
        return render_automation_design(spec, prompt)
    if spec.document_type == "tool_bridge_design":
        return render_tool_bridge_design(spec)
    if spec.document_type == "computer_use_workflow":
        return render_computer_use_workflow(spec)
    if spec.document_type == "release_checklist":
        return render_release_checklist(spec)
    if spec.document_type == "invoice":
        return render_invoice(spec)
    if spec.document_type == "proposal":
        return render_proposal(spec)
    if spec.document_type == "launch_plan":
        return render_launch_plan(spec)
    if spec.document_type == "email_sequence":
        return render_email_sequence(spec)
    return render_business_brief(spec)


def validate_markdown(rendered: str, spec: BusinessDocSpec | None = None) -> list[str]:
    lower = rendered.lower()
    errors: list[str] = []
    if len(rendered.strip()) < 650:
        errors.append("document too short")
    if "lorem ipsum" in lower:
        errors.append("lorem ipsum found")
    if "{{" in rendered or "}}" in rendered:
        errors.append("template placeholders found")
    if not rendered.lstrip().startswith("# "):
        errors.append("missing title")
    if spec:
        if spec.document_type not in TECHNICAL_DOCUMENT_TYPES and spec.business_name.lower() not in lower:
            errors.append("missing business name")
        if spec.document_type == "invoice":
            for token in ["total", "due", "payment", "services"]:
                if token not in lower:
                    errors.append(f"missing invoice token: {token}")
        elif spec.document_type == "proposal":
            for token in ["scope", "timeline", "investment", "approval"]:
                if token not in lower:
                    errors.append(f"missing proposal token: {token}")
        elif spec.document_type == "launch_plan":
            for token in ["channels", "schedule", "metrics", "positioning"]:
                if token not in lower:
                    errors.append(f"missing launch token: {token}")
        elif spec.document_type == "email_sequence":
            if lower.count("subject:") < 3:
                errors.append("missing three email subjects")
            if "call to action" not in lower and "reply" not in lower:
                errors.append("missing email CTA")
        elif spec.document_type == "implementation_plan":
            for token in ["file structure", "shell code", "checksum", "atomic", "test"]:
                if token not in lower:
                    errors.append(f"missing implementation token: {token}")
        elif spec.document_type == "debug_runbook":
            for token in ["root causes", "patch plan", "verification checklist", "rollback"]:
                if token not in lower:
                    errors.append(f"missing debug token: {token}")
        elif spec.document_type == "automation_design":
            for token in ["data model", "request lifecycle", "pseudo-code", "rate limit", "failure handling"]:
                if token not in lower:
                    errors.append(f"missing automation token: {token}")
        elif spec.document_type == "tool_bridge_design":
            for token in ["state model", "20-30 seconds", "duplicate", "final summary"]:
                if token not in lower:
                    errors.append(f"missing bridge token: {token}")
        elif spec.document_type == "computer_use_workflow":
            for token in ["browser workflow", "screenshot", "login", "final url", "failure recovery"]:
                if token not in lower:
                    errors.append(f"missing computer-use token: {token}")
        elif spec.document_type == "release_checklist":
            for token in ["dmg rebuild", "signing", "notarization", "rollback", "customer install"]:
                if token not in lower:
                    errors.append(f"missing release token: {token}")
    return errors


def render_from_prompt(prompt: str) -> tuple[BusinessDocSpec, str, list[str]]:
    spec = spec_from_prompt(prompt)
    rendered = render_markdown(spec, prompt)
    return spec, rendered, validate_markdown(rendered, spec)


def write_markdown(path: Path, rendered: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(rendered, encoding="utf-8")


def spec_to_json(spec: BusinessDocSpec) -> str:
    return json.dumps(spec.__dict__, indent=2, ensure_ascii=False)