# PrintNode-clone WebUI + direct :9100 socket — 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:** Give printpapi a PrintNode-shaped web UI (sidebar nav, print-from-browser, devices, history, API-key management, downloads) and let the agent print straight to a network printer by IP (`socket://host:port`).
**Architecture:** Two independent parts. **Part A** (Tasks 1-3) extends the agent so a printer entry can be a raw TCP socket target; the agent opens the socket — poll model stays intact, works behind NAT. **Part B** (Tasks 4-8) adds a `pdf_uri` content type, a `jobs.title` column + `agent_name` in history, and replaces the inline dashboard with a static multi-view SPA served from `app/dashboard.html`.
**Tech Stack:** Python stdlib only (`http.server`, `sqlite3`, `socket`, `urllib`, `pathlib`); pytest; plain HTML/CSS/JS (no framework, no build step).
## Global Constraints
- **Server: Python stdlib only.** No web framework, no new dependencies (CLAUDE.md).
- **Agent: cross-platform**, stdlib `socket` for the network path.
- **Tests: pytest**, no fixtures/frameworks beyond it. Real loopback `ThreadingHTTPServer` + real SQLite (`:memory:` or tempfile), injected fns — no mocks. Run `python -m pytest` from repo root.
- **Security stays** (CLAUDE.md trust-boundary rules): bearer/API-key auth on every endpoint, `hmac.compare_digest`, `http(s)`-only URL fetch, no `shell=True`, temp-file cleanup. Do not regress any of these.
- **Gotcha #1:** never send PDF bytes to a raw/label/socket printer. A `socket://` printer is forced `can_pdf=False`; `pdf` mode to a socket printer raises.
- **No license headers** — existing source files carry none; match the repo.
- TDD, frequent commits. Branch: `feat/printnode-clone-webui` (already checked out).
---
## Part A — Direct network-printer output (agent)
### Task 1: `parse_printers` gains a `target` field
**Files:**
- Modify: `agent/print_agent.py:57-66` (`parse_printers`)
- Test: `agent/tests/test_agent.py`
**Interfaces:**
- Produces: `parse_printers(spec) -> [{"name": str, "can_pdf": bool, "target": str}]`. `target` defaults to `name`; a `socket://host:port` target forces `can_pdf=False`.
- [ ] **Step 1: Update the existing parse test for the new `target` key**
In `agent/tests/test_agent.py`, replace `test_parse_printers_pdf_is_opt_in_default_raw` with:
```python
def test_parse_printers_pdf_is_opt_in_default_raw():
ps = print_agent.parse_printers("Zebra GK420d; HP LaserJet|pdf ; Office|PDF; ")
# default is raw-only so a label printer is never auto-sent a PDF (gotcha #1);
# a document printer opts into PDF with a '|pdf' tag (case-insensitive).
assert ps == [
{"name": "Zebra GK420d", "can_pdf": False, "target": "Zebra GK420d"},
{"name": "HP LaserJet", "can_pdf": True, "target": "HP LaserJet"},
{"name": "Office", "can_pdf": True, "target": "Office"},
]
```
- [ ] **Step 2: Add the failing socket-grammar tests**
Append to `agent/tests/test_agent.py`:
```python
def test_parse_printers_socket_target():
ps = print_agent.parse_printers("Bixolon ; netz = socket://10.0.0.5:9100")
assert ps == [
{"name": "Bixolon", "can_pdf": False, "target": "Bixolon"},
{"name": "netz", "can_pdf": False, "target": "socket://10.0.0.5:9100"},
]
def test_parse_printers_socket_is_forced_raw_only():
# |pdf on a socket target is ignored — no renderer behind a bare socket (gotcha #1)
ps = print_agent.parse_printers("lbl|pdf = socket://10.0.0.5:9100")
assert ps == [{"name": "lbl", "can_pdf": False, "target": "socket://10.0.0.5:9100"}]
```
- [ ] **Step 3: Run the tests to verify they fail**
Run: `python -m pytest agent/tests/test_agent.py -k parse_printers -v`
Expected: FAIL (`test_parse_printers_pdf_is_opt_in_default_raw` KeyError/mismatch on `target`; socket tests fail — no `target` produced).
- [ ] **Step 4: Implement the new `parse_printers`**
Replace `parse_printers` in `agent/print_agent.py` with:
```python
def parse_printers(spec):
"""agent.ini 'printers' (semicolon-separated) -> [{name, can_pdf, target}].
Grammar: name [|pdf] [= target].
- no '=' -> target is the name (a CUPS queue / Windows printer).
- '= socket://host:port' -> agent opens a raw TCP socket to it.
Append '|pdf' to declare a document printer PDF-capable; default is raw-only so a label
printer is never auto-sent a PDF (gotcha #1). A socket:// target is always raw-only."""
out = []
for entry in spec.split(";"):
left, _, target = entry.partition("=")
name, _, tag = left.strip().partition("|")
name = name.strip()
if not name:
continue
target = target.strip() or name
can_pdf = tag.strip().lower() == "pdf"
if target.startswith("socket://"):
can_pdf = False # gotcha #1: no renderer behind a bare socket
out.append({"name": name, "can_pdf": can_pdf, "target": target})
return out
```
- [ ] **Step 5: Run the tests to verify they pass**
Run: `python -m pytest agent/tests/test_agent.py -k parse_printers -v`
Expected: PASS (3 tests).
- [ ] **Step 6: Commit**
```bash
git add agent/print_agent.py agent/tests/test_agent.py
git commit -m "feat(agent): parse_printers supports socket:// targets (raw-only)"
```
---
### Task 2: `raw_to_socket` sender
**Files:**
- Modify: `agent/print_agent.py` (add `import socket` at top; add `raw_to_socket`)
- Test: `agent/tests/test_agent.py`
**Interfaces:**
- Produces: `raw_to_socket(target, data, connect=socket.create_connection, timeout=30)` — parses `socket://host:port`, opens the connection, `sendall(data)`. `connect` is injectable for tests.
- [ ] **Step 1: Write the failing test**
Append to `agent/tests/test_agent.py`:
```python
def test_raw_to_socket_parses_addr_and_sends_bytes():
seen = {}
class FakeSock:
def __enter__(self): return self
def __exit__(self, *a): return False
def sendall(self, d): seen["data"] = d
def fake_connect(addr, timeout=None):
seen["addr"] = addr
return FakeSock()
print_agent.raw_to_socket("socket://10.0.0.5:9100", b"^XA^XZ", connect=fake_connect)
assert seen["addr"] == ("10.0.0.5", 9100)
assert seen["data"] == b"^XA^XZ"
```
- [ ] **Step 2: Run test to verify it fails**
Run: `python -m pytest agent/tests/test_agent.py::test_raw_to_socket_parses_addr_and_sends_bytes -v`
Expected: FAIL with `AttributeError: module ... has no attribute 'raw_to_socket'`.
- [ ] **Step 3: Implement**
Add `import socket` to the imports at the top of `agent/print_agent.py`, then add near the other `*_to_printer*` functions:
```python
def raw_to_socket(target, data, connect=socket.create_connection, timeout=30):
# Raw bytes straight to a network printer's socket (e.g. Zebra :9100). Already-rendered only
# (ZPL/ESC-POS) — gotcha #1: a bare socket has no renderer, so PDF must never reach here.
addr = target[len("socket://"):] if target.startswith("socket://") else target
host, _, port = addr.rpartition(":") # ponytail: IPv4 host:port; no IPv6-in-brackets support
with connect((host, int(port)), timeout=timeout) as s:
s.sendall(data)
```
- [ ] **Step 4: Run test to verify it passes**
Run: `python -m pytest agent/tests/test_agent.py::test_raw_to_socket_parses_addr_and_sends_bytes -v`
Expected: PASS.
- [ ] **Step 5: Commit**
```bash
git add agent/print_agent.py agent/tests/test_agent.py
git commit -m "feat(agent): raw_to_socket sends bytes to a network printer socket"
```
---
### Task 3: `print_job` routes socket vs local; wire `run_once`/`main`
**Files:**
- Modify: `agent/print_agent.py` (`print_job`, `main`)
- Test: `agent/tests/test_agent.py` (new socket-routing tests; update `test_print_job_bad_mode_raises` and the three `run_once` tests to pass printer **entries**)
**Interfaces:**
- Consumes: entries from `parse_printers` (Task 1); `raw_to_socket` (Task 2).
- Produces: `print_job(mode, entry, data, raw_fn=raw_to_printer, pdf_fn=pdf_to_printer, socket_fn=raw_to_socket)` — `entry` is a `{"name","can_pdf","target"}` dict. `run_once`'s `printer_by_id` now maps `job_id`→**entry** (not a name string).
- [ ] **Step 1: Update the existing tests that pass a printer as a bare string**
In `agent/tests/test_agent.py`:
Replace `test_print_job_bad_mode_raises` with:
```python
def test_print_job_bad_mode_raises():
entry = {"name": "P", "can_pdf": False, "target": "P"}
with pytest.raises(ValueError, match="bad mode"):
print_agent.print_job("docx", entry, b"x")
```
In the three `run_once` tests, change the `printer_by_id` argument from `{1: "Zebra"}` to an entry dict `{1: {"name": "Zebra", "can_pdf": False, "target": "Zebra"}}`. Concretely:
- `test_run_once_prints_and_reports_success`: `"http://x", "k", {1: {"name": "Zebra", "can_pdf": False, "target": "Zebra"}},`
- `test_run_once_reports_failure_on_print_error`: same `{1: {"name": "Zebra", "can_pdf": False, "target": "Zebra"}}`
- `test_run_once_prints_pdf_and_reports_success`: `{1: {"name": "Zebra", "can_pdf": True, "target": "Zebra"}}`
(The `raw_fn`/`pdf_fn` lambdas receive `entry["target"]` == `"Zebra"`, so the `printed == {"printer": "Zebra", ...}` assertions still hold. `test_run_once_no_job_returns_false` keeps `{}`.)
- [ ] **Step 2: Add the failing socket-routing tests**
Append to `agent/tests/test_agent.py`:
```python
def test_print_job_socket_raw_routes_to_socket_fn():
seen = {}
entry = {"name": "netz", "can_pdf": False, "target": "socket://10.0.0.5:9100"}
print_agent.print_job(
"raw", entry, b"Z",
socket_fn=lambda t, d: seen.update(t=t, d=d),
raw_fn=lambda *a: (_ for _ in ()).throw(AssertionError("local raw not expected")),
pdf_fn=lambda *a: None)
assert seen == {"t": "socket://10.0.0.5:9100", "d": b"Z"}
def test_print_job_socket_pdf_is_refused():
entry = {"name": "netz", "can_pdf": False, "target": "socket://10.0.0.5:9100"}
with pytest.raises(ValueError, match="raw-only"):
print_agent.print_job("pdf", entry, b"%PDF",
socket_fn=lambda t, d: None,
raw_fn=lambda *a: None, pdf_fn=lambda *a: None)
def test_print_job_local_raw_uses_target_name():
seen = {}
entry = {"name": "Zebra", "can_pdf": False, "target": "Zebra"}
print_agent.print_job("raw", entry, b"x",
raw_fn=lambda t, d: seen.update(t=t, d=d), pdf_fn=lambda *a: None)
assert seen == {"t": "Zebra", "d": b"x"}
```
- [ ] **Step 3: Run tests to verify they fail**
Run: `python -m pytest agent/tests/test_agent.py -v`
Expected: the socket-routing tests FAIL (`print_job` still expects a string printer / has no `socket_fn`).
- [ ] **Step 4: Implement `print_job` routing and wire `main`**
Replace `print_job` in `agent/print_agent.py` with:
```python
def print_job(mode, entry, data, raw_fn=raw_to_printer, pdf_fn=pdf_to_printer,
socket_fn=raw_to_socket):
target = entry["target"]
if target.startswith("socket://"):
if mode != "raw":
raise ValueError("network socket printer is raw-only (cannot render PDF)")
socket_fn(target, data)
return
if mode == "raw":
raw_fn(target, data)
elif mode == "pdf":
pdf_fn(target, data)
else:
raise ValueError(f"bad mode: {mode}")
```
`run_once` needs no code change — it already does `printer = printer_by_id.get(job["printer_id"])` and passes it to `print_job`; that value is now an entry dict.
In `main()`, change how `printer_by_id` is built so it maps id→entry:
```python
printers = parse_printers(cfg["agent"]["printers"])
reg = register(base, key, name, printers)
entry_by_name = {p["name"]: p for p in printers}
printer_by_id = {pid: entry_by_name[pname] for pname, pid in reg["printer_ids"].items()}
print(f"print-agent registered as computer {reg['computer_id']}, printers={printer_by_id}")
```
(The `register` call still sends the full entry dicts; the server reads only `name`/`can_pdf` and ignores `target`.)
- [ ] **Step 5: Run the whole agent suite**
Run: `python -m pytest agent/tests/test_agent.py -v`
Expected: PASS (all, including the updated `run_once`/`bad_mode` tests and the 3 new socket tests).
- [ ] **Step 6: Commit**
```bash
git add agent/print_agent.py agent/tests/test_agent.py
git commit -m "feat(agent): print_job routes socket:// printers; run_once carries entries"
```
---
## Part B — WebUI
### Task 4: `pdf_uri` content type
**Files:**
- Modify: `app/dispatch.py:54-59` (`raw_uri` branch) and `agent_mode`
- Test: `tests/test_dispatch.py`
**Interfaces:**
- Produces: `type: "pdf_uri"` — server GETs `url` and prints the bytes as PDF. `agent_mode("pdf_uri") == "pdf"`.
- [ ] **Step 1: Write the failing test**
Append to `tests/test_dispatch.py`:
```python
def test_pdf_uri_uses_injected_fetcher_and_is_pdf_mode():
seen = {}
def fake(url):
seen["url"] = url
return b"%PDF-bytes"
assert decode_payload({"type": "pdf_uri", "url": "https://x/a.pdf"}, fetch_url=fake) == b"%PDF-bytes"
assert seen["url"] == "https://x/a.pdf"
assert agent_mode("pdf_uri") == "pdf"
```
- [ ] **Step 2: Run test to verify it fails**
Run: `python -m pytest tests/test_dispatch.py::test_pdf_uri_uses_injected_fetcher_and_is_pdf_mode -v`
Expected: FAIL (`DispatchError: unknown type: 'pdf_uri'`).
- [ ] **Step 3: Implement**
In `app/dispatch.py`, change the `raw_uri` branch to accept `pdf_uri` too:
```python
if t in ("raw_uri", "pdf_uri"):
url = _checked_http_url(body.get("url"))
try:
return fetch_url(url)
except Exception as e:
raise FetchError(f"fetch failed: {e}") from e
```
And extend `agent_mode`:
```python
def agent_mode(type_):
return "pdf" if type_ in ("pdf_base64", "pdf_uri_post", "pdf_uri") else "raw"
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `python -m pytest tests/test_dispatch.py -v`
Expected: PASS (new test + all existing dispatch tests).
- [ ] **Step 5: Commit**
```bash
git add app/dispatch.py tests/test_dispatch.py
git commit -m "feat: add pdf_uri content type (GET a URL, print as PDF)"
```
---
### Task 5: `jobs.title` column + `agent_name` in history
**Files:**
- Modify: `app/store.py` (`_SCHEMA` jobs table; `init_db` migration; `enqueue_job`; `recent_jobs`)
- Test: `tests/test_store.py`
**Interfaces:**
- Produces: `enqueue_job(conn, printer_id, type_, mode, payload, user_id=DEFAULT_USER, title=None)`; `recent_jobs` rows gain `title` and `agent_name`.
- [ ] **Step 1: Write the failing tests**
Append to `tests/test_store.py`:
```python
def test_enqueue_and_recent_jobs_carry_title_and_agent_name():
conn = _db()
reg = store.register_agent(conn, "pc-1", "k", [{"name": "Z", "can_pdf": False}])
pid = reg["printer_ids"]["Z"]
store.enqueue_job(conn, pid, "raw_base64", "raw", b"x", title="Versandlabel")
row = store.recent_jobs(conn)[0]
assert row["title"] == "Versandlabel"
assert row["agent_name"] == "pc-1"
assert row["printer_name"] == "Z"
def test_title_column_migration_is_idempotent():
conn = _db()
store.init_db(conn) # run migration a second time — must not raise or duplicate
cols = [r["name"] for r in conn.execute("PRAGMA table_info(jobs)").fetchall()]
assert cols.count("title") == 1
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `python -m pytest tests/test_store.py -k "title or agent_name" -v`
Expected: FAIL (`enqueue_job` has no `title` kwarg / `recent_jobs` lacks `title`+`agent_name`).
- [ ] **Step 3: Add `title` to the schema and an idempotent migration**
In `app/store.py`, add `title TEXT` to the `jobs` CREATE (self-documenting for fresh DBs):
```python
CREATE TABLE IF NOT EXISTS jobs(
id INTEGER PRIMARY KEY AUTOINCREMENT, org_id INTEGER NOT NULL, user_id INTEGER NOT NULL,
printer_id INTEGER NOT NULL, agent_id INTEGER NOT NULL, type TEXT NOT NULL, mode TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'queued', payload BLOB NOT NULL, error TEXT, title TEXT,
retries INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, claimed_at REAL, finished_at REAL);
```
In `init_db`, right after `conn.executescript(_SCHEMA)` (still inside the `with _LOCK: try:` block), add the migration for DBs created before this change:
```python
conn.executescript(_SCHEMA)
try:
conn.execute("ALTER TABLE jobs ADD COLUMN title TEXT")
except sqlite3.OperationalError:
pass # column already exists (fresh DB or prior run) — idempotent
```
- [ ] **Step 4: Thread `title` through `enqueue_job` and `recent_jobs`**
`enqueue_job` — add the `title` param and column:
```python
def enqueue_job(conn, printer_id, type_, mode, payload, user_id=DEFAULT_USER, title=None):
now = time.time()
with _LOCK:
try:
p = conn.execute("SELECT org_id, agent_id FROM printers WHERE id=?",
(printer_id,)).fetchone()
if p is None:
raise UnknownPrinter(f"unknown printer: {printer_id}")
cur = conn.execute(
"INSERT INTO jobs(org_id, user_id, printer_id, agent_id, type, mode, state, "
"payload, title, created_at) VALUES(?,?,?,?,?,?,?,?,?,?)",
(p["org_id"], user_id, printer_id, p["agent_id"], type_, mode, "queued",
sqlite3.Binary(payload), title, now))
conn.commit()
return cur.lastrowid
except Exception:
conn.rollback()
raise
```
`recent_jobs` — join `agents` and return `agent_name` + `title`:
```python
def recent_jobs(conn, limit=50):
with _LOCK:
rows = conn.execute(
"SELECT j.id, j.printer_id, p.name AS printer_name, a.name AS agent_name, j.title, "
"j.state, j.type, j.mode, j.error, j.created_at, j.finished_at "
"FROM jobs j JOIN printers p ON p.id = j.printer_id "
"JOIN agents a ON a.id = j.agent_id "
"ORDER BY j.id DESC LIMIT ?", (limit,)).fetchall()
return [dict(r) for r in rows]
```
- [ ] **Step 5: Run the store suite**
Run: `python -m pytest tests/test_store.py -v`
Expected: PASS (new tests + existing `test_recent_jobs_*` and `test_init_seeds_*` still green).
- [ ] **Step 6: Commit**
```bash
git add app/store.py tests/test_store.py
git commit -m "feat: job title column (idempotent migration) + agent_name in history"
```
---
### Task 6: `POST /jobs` accepts `title`
**Files:**
- Modify: `app/server.py` (the `POST /jobs` branch, ~line 304)
- Test: `tests/test_server.py`
**Interfaces:**
- Consumes: `store.enqueue_job(..., title=...)` (Task 5).
- [ ] **Step 1: Write the failing test**
Append to `tests/test_server.py`:
```python
def test_job_title_roundtrips_through_http():
conn = _mem()
reg = store.register_agent(conn, "pc", "ak", [{"name": "Z", "can_pdf": False}])
pid = reg["printer_ids"]["Z"]
httpd, base = _serve(conn)
try:
code, _ = _req("POST", base + "/jobs", token="t",
body={"printer_id": pid, "type": "raw_base64",
"content": "QUJD", "title": "Label #5"})
assert code == 200
jobs = json.loads(_req("GET", base + "/jobs", token="t")[1])["jobs"]
assert jobs[0]["title"] == "Label #5"
assert jobs[0]["agent_name"] == "pc"
finally:
httpd.shutdown()
```
- [ ] **Step 2: Run test to verify it fails**
Run: `python -m pytest tests/test_server.py::test_job_title_roundtrips_through_http -v`
Expected: FAIL (`title` is `None` in the response — server doesn't pass it).
- [ ] **Step 3: Implement**
In `app/server.py`, in the `POST /jobs` branch, pass the title to `enqueue_job`:
```python
jid = store.enqueue_job(conn, body.get("printer_id"), body.get("type"),
mode, data, title=body.get("title"))
```
- [ ] **Step 4: Run test to verify it passes**
Run: `python -m pytest tests/test_server.py -v`
Expected: PASS (new test + all existing server tests).
- [ ] **Step 5: Commit**
```bash
git add app/server.py tests/test_server.py
git commit -m "feat: POST /jobs stores an optional job title"
```
---
### Task 7: Create `app/dashboard.html` (multi-view SPA)
**Files:**
- Create: `app/dashboard.html`
**Interfaces:**
- Consumes (client-side, over HTTP): `GET /printers`, `GET /jobs`, `POST /jobs`, `GET/POST /apikeys`, `DELETE /apikeys/{id}`. Uses content types `pdf_base64`/`raw_base64`/`raw_uri`/`pdf_uri` and the `title` field (Tasks 4-6).
- Produces: the static asset `app/server.py` loads in Task 8. Placeholders `__PDF_B64__` / `__ZPL_B64__` are substituted at load time. Must contain the literal substrings the existing `test_dashboard_served_at_root_as_html` asserts: ``, `printpapi`, `/printers`, `/jobs`.
- [ ] **Step 1: Create the file with this exact content**
Create `app/dashboard.html`:
```html
printpapi
printpapi
Enter your API token to connect.
Print Something
```
- [ ] **Step 2: Sanity-check the asset exists and has the required markers**
Run: `python -c "import pathlib,sys; h=pathlib.Path('app/dashboard.html').read_text(); [sys.exit('missing '+m) for m in ('','printpapi','/printers','/jobs','#downloads') if m not in h]; print('ok')"`
Expected: `ok`.
- [ ] **Step 3: Commit**
```bash
git add app/dashboard.html
git commit -m "feat(ui): PrintNode-style dashboard SPA (print, devices, history, keys, downloads)"
```
---
### Task 8: Serve `dashboard.html` from `app/server.py`
**Files:**
- Modify: `app/server.py` (add `pathlib` import; replace the inline `_DASHBOARD_HTML` string with a file read; keep the `_TEST_PDF_B64` / `_TEST_ZPL_B64` constants and the `.replace(...)` substitution)
- Test: `tests/test_server.py`
**Interfaces:**
- Consumes: `app/dashboard.html` (Task 7).
- [ ] **Step 1: Write the failing test**
Append to `tests/test_server.py`:
```python
def test_dashboard_has_sidebar_nav():
conn = _mem()
httpd, base = _serve(conn)
try:
html = _req("GET", base + "/")[1].decode()
for h in ("#print", "#devices", "#history", "#keys", "#downloads"):
assert h in html
assert "Sign Out" in html
finally:
httpd.shutdown()
```
- [ ] **Step 2: Run test to verify it fails**
Run: `python -m pytest tests/test_server.py::test_dashboard_has_sidebar_nav -v`
Expected: FAIL (the old inline dashboard has no sidebar `#print`/`#devices`/… nav).
- [ ] **Step 3: Replace the inline HTML with a file read**
In `app/server.py`: add `from pathlib import Path` to the imports. Keep the `_TEST_PDF_B64` and `_TEST_ZPL_B64` constants (and the comment above them). **Delete** the entire inline `_DASHBOARD_HTML = r"""…"""` block (the ` …