Levandor

Knowledgebase — implementation plan

18 min read

Knowledgebase Implementation Plan

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

Goal: A tailnet-only service on telep-mainframe that browses/searches OCR’d PDF manuals (mkdocs-material static site) and ingests new PDFs via an upload portal (OCR → split → rebuild).

Architecture: A Flask+waitress app serves an mkdocs-built static site at / and hosts an upload portal that runs marker OCR (reusing ~/ocr/venv) on a background GPU-serialised worker, splits the markdown into per-section pages under a browsable docs/ tree, and atomically rebuilds the site.

Tech Stack: Python 3.13 venv (flask, waitress, mkdocs-material), the existing ~/ocr marker pipeline, systemd, tailscale serve.

Design: 2026-07-24-knowledgebase-design

Global Constraints

  • All code + data under /home/levander/knowledgebase/ on telep-mainframe. App runs as levander (NOT root).
  • No comments, docstrings, or inline annotations in any code (user’s global rule). Self-explanatory naming only.
  • Do not run git commit. Each task ends in a verification step; commit steps omitted per the user’s rule.
  • A dedicated venv at /home/levander/knowledgebase/venv (flask, waitress, mkdocs-material). The OCR venv /home/levander/ocr/venv is REUSED only as a subprocess (marker_single), never imported.
  • Service/site name is knowledgebase; site title “Knowledgebase”.
  • Security: folder + manual name from uploads MUST pass safe_slug and the destination MUST be realpath-verified inside docs/ before any filesystem write. Uploads accepted only if the content starts with the bytes %PDF.
  • Library source lives at manuals-src/docs/<folder>/<name>/; the mkdocs build output is site/ (served static).
  • The app binds 127.0.0.1:8092; tailscale serve :8446 maps to it.
  • Run commands on the box: /usr/bin/ssh levander@telep-mainframe. On this Mac bare ssh is broken — always /usr/bin/ssh. Use sudo only for the systemd unit + tailscale serve in Task 7.

Test command (every task): cd /home/levander/knowledgebase && venv/bin/python -m unittest <module> -v

Task 1: split_manual.py — split a marker markdown into per-section pages

Files:

  • Create: /home/levander/knowledgebase/split_manual.py
  • Test: /home/levander/knowledgebase/test_split_manual.py

Interfaces:

  • Produces:

    • slugify(text) -> str — lowercase, [a-z0-9-], collapsed, trimmed, ≤50 chars, "section" if empty
    • split_manual(markdown, title) -> list[dict] — each {"order": int, "slug": str, "heading": str, "body": str}; splits on ^# (h1); content before the first h1 becomes an Introduction section; slug is a 3-digit zero-padded order prefix + slugified heading; body includes the heading line and a trailing newline
    • write_sections(sections, dest_dir) -> list[str] — writes <slug>.md files, returns the filenames

  • Step 1: Create the project dir + venv

/usr/bin/ssh levander@telep-mainframe 'mkdir -p /home/levander/knowledgebase && cd /home/levander/knowledgebase && python3 -m venv venv && venv/bin/pip install -q --upgrade pip && echo "venv ready"'
  • Step 2: Write the failing test

Create /home/levander/knowledgebase/test_split_manual.py:

import unittest
 
from split_manual import slugify, split_manual, write_sections
 
PREAMBLE = """http://example
Some front matter.
 
![](_page_0_Picture_2.jpeg)
 
# Engine Mechanical
 
Torque the bolts.
 
![](_page_5_Figure_1.jpeg)
 
# Body & Trim
 
Panel gaps.
"""
 
NO_PREAMBLE = """# First
 
a
 
# Second
 
b
"""
 
 
class TestSlugify(unittest.TestCase):
    def test_basic(self):
        self.assertEqual(slugify("Engine Mechanical"), "engine-mechanical")
 
    def test_symbols_collapse(self):
        self.assertEqual(slugify("Body & Trim!!"), "body-trim")
 
    def test_empty_fallback(self):
        self.assertEqual(slugify("###"), "section")
 
    def test_truncated(self):
        self.assertLessEqual(len(slugify("x" * 200)), 50)
 
 
class TestSplit(unittest.TestCase):
    def test_preamble_becomes_intro(self):
        sections = split_manual(PREAMBLE, "Vitara")
        self.assertEqual(sections[0]["heading"], "Introduction")
        self.assertEqual(sections[0]["order"], 1)
        self.assertEqual(sections[0]["slug"], "001-introduction")
 
    def test_section_count_and_order(self):
        sections = split_manual(PREAMBLE, "Vitara")
        self.assertEqual(len(sections), 3)
        self.assertEqual([s["order"] for s in sections], [1, 2, 3])
        self.assertEqual(sections[1]["slug"], "002-engine-mechanical")
 
    def test_no_preamble_no_intro(self):
        sections = split_manual(NO_PREAMBLE, "X")
        self.assertEqual(len(sections), 2)
        self.assertEqual(sections[0]["heading"], "First")
 
    def test_image_refs_preserved(self):
        sections = split_manual(PREAMBLE, "Vitara")
        self.assertIn("![](_page_5_Figure_1.jpeg)", sections[1]["body"])
 
    def test_intro_body_has_heading(self):
        sections = split_manual(PREAMBLE, "Vitara")
        self.assertTrue(sections[0]["body"].startswith("# Introduction"))
 
    def test_body_keeps_own_heading(self):
        sections = split_manual(NO_PREAMBLE, "X")
        self.assertTrue(sections[0]["body"].startswith("# First"))
 
    def test_empty_input(self):
        self.assertEqual(split_manual("", "X"), [])
 
 
class TestWrite(unittest.TestCase):
    def test_writes_files(self):
        import tempfile, os
        sections = split_manual(NO_PREAMBLE, "X")
        d = tempfile.mkdtemp()
        names = write_sections(sections, d)
        self.assertEqual(names, ["001-first.md", "002-second.md"])
        self.assertTrue(os.path.exists(os.path.join(d, "001-first.md")))
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 3: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest test_split_manual -v'

Expected: FAIL with ModuleNotFoundError: No module named 'split_manual'

  • Step 4: Write the implementation

Create /home/levander/knowledgebase/split_manual.py:

import os
import re
 
_SLUG = re.compile(r"[^a-z0-9]+")
 
 
def slugify(text):
    slug = _SLUG.sub("-", text.strip().lower()).strip("-")
    slug = slug[:50].strip("-")
    return slug or "section"
 
 
def split_manual(markdown, title):
    if not markdown.strip():
        return []
    blocks = []
    heading = None
    body = []
    for line in markdown.splitlines():
        if line.startswith("# "):
            if heading is not None or body:
                blocks.append((heading, body))
            heading = line[2:].strip()
            body = [line]
        else:
            body.append(line)
    if heading is not None or body:
        blocks.append((heading, body))
    sections = []
    order = 0
    for block_heading, block_lines in blocks:
        text = "\n".join(block_lines).strip()
        if block_heading is None:
            if not text:
                continue
            block_heading = "Introduction"
            text = "# Introduction\n\n" + text
        order += 1
        sections.append(
            {
                "order": order,
                "slug": "%03d-%s" % (order, slugify(block_heading)),
                "heading": block_heading,
                "body": text + "\n",
            }
        )
    return sections
 
 
def write_sections(sections, dest_dir):
    names = []
    for section in sections:
        name = section["slug"] + ".md"
        with open(os.path.join(dest_dir, name), "w", encoding="utf-8") as handle:
            handle.write(section["body"])
        names.append(name)
    return names
  • Step 5: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest test_split_manual -v'

Expected: OK — 12 tests pass.

  • Step 6: Verify against the real Vitara markdown
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -c "
from split_manual import split_manual
md = open(\"/home/levander/ocr/marker_out/g16b_manual_specific/g16b_manual_specific.md\", encoding=\"utf-8\").read()
s = split_manual(md, \"Vitara\")
print(len(s), \"sections\")
print(s[0][\"slug\"], \"|\", s[1][\"slug\"])
"'

Expected: a section count around 58-59 and sensible slugs.

Task 2: slugs.py — safe slug + path-traversal guard

Files:

  • Create: /home/levander/knowledgebase/slugs.py
  • Test: /home/levander/knowledgebase/test_slugs.py

Interfaces:

  • Produces:

    • safe_slug(text) -> str — lowercase [a-z0-9-], collapsed, ≤60 chars; raises ValueError if the result is empty
    • resolve_within(base_dir, *parts) -> str — realpath of base_dir/parts...; raises ValueError if it escapes base_dir

  • Step 1: Write the failing test

Create /home/levander/knowledgebase/test_slugs.py:

import os
import tempfile
import unittest
 
from slugs import resolve_within, safe_slug
 
 
class TestSafeSlug(unittest.TestCase):
    def test_spaces_to_hyphens(self):
        self.assertEqual(safe_slug("Engine Mechanical"), "engine-mechanical")
 
    def test_traversal_stripped(self):
        self.assertEqual(safe_slug("../../etc"), "etc")
 
    def test_only_symbols_raises(self):
        with self.assertRaises(ValueError):
            safe_slug("../../")
 
    def test_empty_raises(self):
        with self.assertRaises(ValueError):
            safe_slug("")
 
    def test_length_capped(self):
        self.assertLessEqual(len(safe_slug("a" * 200)), 60)
 
 
class TestResolveWithin(unittest.TestCase):
    def setUp(self):
        self.base = tempfile.mkdtemp()
 
    def test_ok_path(self):
        target = resolve_within(self.base, "chevy", "owners")
        self.assertTrue(target.startswith(os.path.realpath(self.base) + os.sep))
 
    def test_escape_raises(self):
        with self.assertRaises(ValueError):
            resolve_within(self.base, "..", "..", "etc")
 
    def test_absolute_component_raises(self):
        with self.assertRaises(ValueError):
            resolve_within(self.base, "/etc", "passwd")
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest test_slugs -v'

Expected: FAIL with ModuleNotFoundError: No module named 'slugs'

  • Step 3: Write the implementation

Create /home/levander/knowledgebase/slugs.py:

import os
import re
 
_SLUG = re.compile(r"[^a-z0-9]+")
 
 
def safe_slug(text):
    slug = _SLUG.sub("-", text.strip().lower()).strip("-")
    slug = slug[:60].strip("-")
    if not slug:
        raise ValueError("empty slug")
    return slug
 
 
def resolve_within(base_dir, *parts):
    base = os.path.realpath(base_dir)
    target = os.path.realpath(os.path.join(base, *parts))
    if target != base and not target.startswith(base + os.sep):
        raise ValueError("path escapes base")
    return target
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest test_slugs -v'

Expected: OK — 8 tests pass.

Task 3: mkdocs scaffold + build.py (atomic swap) + lazy-image hook

Files:

  • Create: /home/levander/knowledgebase/mkdocs.yml
  • Create: /home/levander/knowledgebase/hooks.py
  • Create: /home/levander/knowledgebase/build.py
  • Create: /home/levander/knowledgebase/manuals-src/docs/.gitkeep (placeholder so docs_dir exists)

Interfaces:

  • Produces: build_site() -> tuple[bool, str] — runs mkdocs build into a temp dir, atomically swaps site/, returns (ok, error_tail)

  • Step 1: Install mkdocs-material into the venv

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/pip install -q mkdocs-material && venv/bin/mkdocs --version'

Expected: prints an mkdocs version.

  • Step 2: Create the mkdocs config, hook, and a placeholder page
/usr/bin/ssh levander@telep-mainframe 'mkdir -p /home/levander/knowledgebase/manuals-src/docs && cat > /home/levander/knowledgebase/mkdocs.yml <<'"'"'YML'"'"'
site_name: Knowledgebase
docs_dir: manuals-src/docs
use_directory_urls: true
theme:
  name: material
  palette:
    - scheme: default
      toggle:
        icon: material/brightness-7
        name: Dark mode
    - scheme: slate
      toggle:
        icon: material/brightness-4
        name: Light mode
  features:
    - navigation.instant
    - navigation.top
    - navigation.sections
    - search.suggest
    - search.highlight
plugins:
  - search
hooks:
  - hooks.py
YML
cat > /home/levander/knowledgebase/hooks.py <<'"'"'PY'"'"'
def on_post_page(output, page, config):
    return output.replace("<img ", "<img loading=\"lazy\" ")
PY
cat > /home/levander/knowledgebase/manuals-src/docs/index.md <<'"'"'MD'"'"'
# Knowledgebase
 
Select a manual from the navigation, or add one via the upload portal.
MD
echo written'
  • Step 3: Write build.py

Create /home/levander/knowledgebase/build.py:

import os
import shutil
import subprocess
import tempfile
 
PROJECT = os.path.dirname(os.path.abspath(__file__))
SITE = os.path.join(PROJECT, "site")
MKDOCS = os.path.join(PROJECT, "venv", "bin", "mkdocs")
 
 
def build_site():
    tmp = tempfile.mkdtemp(prefix="kb-site-", dir=PROJECT)
    try:
        result = subprocess.run(
            [MKDOCS, "build", "--site-dir", tmp, "--quiet"],
            cwd=PROJECT,
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            shutil.rmtree(tmp, ignore_errors=True)
            return False, (result.stderr or result.stdout)[-2000:]
        backup = SITE + ".old"
        shutil.rmtree(backup, ignore_errors=True)
        if os.path.exists(SITE):
            os.rename(SITE, backup)
        os.rename(tmp, SITE)
        shutil.rmtree(backup, ignore_errors=True)
        return True, ""
    except Exception as error:
        shutil.rmtree(tmp, ignore_errors=True)
        return False, str(error)
  • Step 4: Build the scaffold site and confirm the atomic swap works
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -c "
from build import build_site
ok, err = build_site()
print(\"build ok:\", ok, err[:200])
import os
print(\"index exists:\", os.path.exists(\"site/index.html\"))
print(\"search index:\", os.path.exists(\"site/search/search_index.json\"))
"'

Expected: build ok: True, index exists: True, search index: True.

  • Step 5: Confirm a broken source does not destroy the served site
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && printf "site_name: x\ndocs_dir: nonexistent-dir\n" > /tmp/bad.yml && cp mkdocs.yml mkdocs.yml.good && cp /tmp/bad.yml mkdocs.yml && venv/bin/python -c "
from build import build_site
ok, err = build_site()
print(\"broken build ok:\", ok)
import os
print(\"old site still present:\", os.path.exists(\"site/index.html\"))
"; cp mkdocs.yml.good mkdocs.yml; rm -f mkdocs.yml.good'

Expected: broken build ok: False and old site still present: True (the previous good site/ was NOT clobbered).

Task 4: seed_import.py — import the two existing OCR’d manuals

Files:

  • Create: /home/levander/knowledgebase/seed_import.py

Interfaces:

  • Consumes: split_manual.split_manual, split_manual.write_sections, slugs.resolve_within

  • Produces: import_manual(src_md, images_dir, folder, name, title) -> str — creates manuals-src/docs/<folder>/<name>/ with split pages + copied images, returns the dest path

  • Step 1: Write seed_import.py

Create /home/levander/knowledgebase/seed_import.py:

import os
import shutil
import sys
 
from slugs import resolve_within, safe_slug
from split_manual import split_manual, write_sections
 
PROJECT = os.path.dirname(os.path.abspath(__file__))
DOCS = os.path.join(PROJECT, "manuals-src", "docs")
IMAGE_SUFFIXES = (".jpeg", ".jpg", ".png")
 
 
def import_manual(src_md, images_dir, folder, name, title):
    dest = resolve_within(DOCS, safe_slug(folder), safe_slug(name))
    os.makedirs(dest, exist_ok=True)
    with open(src_md, encoding="utf-8") as handle:
        markdown = handle.read()
    write_sections(split_manual(markdown, title), dest)
    for entry in os.listdir(images_dir):
        if entry.lower().endswith(IMAGE_SUFFIXES):
            shutil.copy2(os.path.join(images_dir, entry), dest)
    return dest
 
 
if __name__ == "__main__":
    print(import_manual(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5]))
  • Step 2: Import the Vitara supplement
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python seed_import.py /home/levander/ocr/marker_out/g16b_manual_specific/g16b_manual_specific.md /home/levander/ocr/marker_out/g16b_manual_specific suzuki-vitara 5door-supplement "Suzuki Vitara 5-Door Supplement"'

Expected: prints the dest path .../docs/suzuki-vitara/5door-supplement.

  • Step 3: Import the Tracker owners manual (if OCR has finished)
/usr/bin/ssh levander@telep-mainframe 'ls /home/levander/ocr/marker_out/tracker_owners_1995/tracker_owners_1995.md 2>/dev/null && cd /home/levander/knowledgebase && venv/bin/python seed_import.py /home/levander/ocr/marker_out/tracker_owners_1995/tracker_owners_1995.md /home/levander/ocr/marker_out/tracker_owners_1995 chevy-tracker owners-1995 "Chevrolet Tracker 1995 Owners Manual" || echo "tracker OCR not done yet - import later"'

Expected: the dest path, or the “not done yet” message.

  • Step 4: Rebuild and verify the manuals render with images and search
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -c "from build import build_site; print(build_site()[0])" && echo "--- pages ---" && ls site/suzuki-vitara/5door-supplement/ | head && echo "--- an image copied? ---" && ls manuals-src/docs/suzuki-vitara/5door-supplement/*.jpeg | wc -l && echo "--- search finds a term ---" && grep -ci "torque" site/search/search_index.json'

Expected: True, a list of section page folders, a non-zero image count, and a non-zero grep count for a known term.

Task 5: ingest.py — background OCR→split→build worker

Files:

  • Create: /home/levander/knowledgebase/ingest.py

Interfaces:

  • Consumes: split_manual, slugs.resolve_within, slugs.safe_slug, build.build_site

  • Produces:

    • submit(pdf_path, folder, name, title) -> str (job id)
    • jobs (dict id → {"state","step","detail"})
    • start_worker() — starts the single daemon worker thread
    • DOCS (the docs dir path)

  • Step 1: Write ingest.py

Create /home/levander/knowledgebase/ingest.py:

import os
import queue
import shutil
import subprocess
import tempfile
import threading
import uuid
 
from build import build_site
from slugs import resolve_within, safe_slug
from split_manual import split_manual, write_sections
 
PROJECT = os.path.dirname(os.path.abspath(__file__))
DOCS = os.path.join(PROJECT, "manuals-src", "docs")
MARKER = "/home/levander/ocr/venv/bin/marker_single"
IMAGE_SUFFIXES = (".jpeg", ".jpg", ".png")
 
jobs = {}
_queue = queue.Queue()
 
 
def submit(pdf_path, folder, name, title):
    job_id = uuid.uuid4().hex[:12]
    jobs[job_id] = {"state": "queued", "step": "queued", "detail": ""}
    _queue.put((job_id, pdf_path, folder, name, title))
    return job_id
 
 
def _process(job_id, pdf_path, folder, name, title):
    work = tempfile.mkdtemp(prefix="kb-ocr-")
    try:
        jobs[job_id].update(state="ocr", step="running OCR (this takes a while)")
        environment = dict(os.environ, TORCH_DEVICE="cuda")
        result = subprocess.run(
            [MARKER, pdf_path, "--output_dir", work, "--force_ocr"],
            capture_output=True,
            text=True,
            env=environment,
        )
        if result.returncode != 0:
            jobs[job_id].update(state="failed", step="OCR failed", detail=result.stderr[-1500:])
            return
        stem = os.path.splitext(os.path.basename(pdf_path))[0]
        out_dir = os.path.join(work, stem)
        md_path = os.path.join(out_dir, stem + ".md")
        if not os.path.exists(md_path):
            jobs[job_id].update(state="failed", step="OCR produced no markdown", detail=out_dir)
            return
        jobs[job_id].update(state="splitting", step="splitting into pages")
        dest = resolve_within(DOCS, safe_slug(folder), safe_slug(name))
        os.makedirs(dest, exist_ok=True)
        with open(md_path, encoding="utf-8") as handle:
            write_sections(split_manual(handle.read(), title), dest)
        for entry in os.listdir(out_dir):
            if entry.lower().endswith(IMAGE_SUFFIXES):
                shutil.copy2(os.path.join(out_dir, entry), dest)
        jobs[job_id].update(state="building", step="rebuilding site")
        ok, error = build_site()
        if not ok:
            jobs[job_id].update(state="failed", step="build failed", detail=error)
            return
        jobs[job_id].update(state="done", step="done", detail=dest)
    except Exception as error:
        jobs[job_id].update(state="failed", step="error", detail=str(error))
    finally:
        shutil.rmtree(work, ignore_errors=True)
        if os.path.exists(pdf_path):
            os.remove(pdf_path)
 
 
def _worker():
    while True:
        args = _queue.get()
        _process(*args)
        _queue.task_done()
 
 
def start_worker():
    thread = threading.Thread(target=_worker, daemon=True)
    thread.start()
  • Step 2: Smoke-test the pipeline on a tiny generated PDF

This exercises OCR→split→build end to end without the portal. Generate a 1-page PDF, submit it, wait for done.

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -c "
import img2pdf" 2>/dev/null || venv/bin/pip -q install img2pdf 2>/dev/null; cd /home/levander/knowledgebase && venv/bin/python -c "
import time, ingest
from reportlab.pdfgen import canvas
" 2>/dev/null; cd /home/levander/knowledgebase && venv/bin/python -c "
import subprocess
subprocess.run([\"bash\",\"-c\",\"printf %s '"'"'%PDF-1.4 test'"'"' > /dev/null\"])
"'

Then run the actual smoke test (uses the marker venv which can OCR any PDF; use the smallest real PDF available — a 1-page slice of an existing manual):

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -c "
import time, os, ingest
ingest.start_worker()
src = \"/home/levander/ocr/tracker_owners_1995.pdf\"
import shutil; shutil.copy2(src, \"/tmp/kbsmoke.pdf\")
jid = ingest.submit(\"/tmp/kbsmoke.pdf\", \"test\", \"smoke\", \"Smoke Test\")
for _ in range(600):
    st = ingest.jobs[jid]
    if st[\"state\"] in (\"done\",\"failed\"): break
    time.sleep(5)
print(\"final:\", ingest.jobs[jid][\"state\"], ingest.jobs[jid][\"step\"])
print(\"dest exists:\", os.path.exists(\"manuals-src/docs/test/smoke\"))
"'

Expected: final: done done and dest exists: True. (This OCRs a full manual so it takes minutes; it proves the whole worker path. Remove the test manual afterward: rm -rf manuals-src/docs/test && venv/bin/python -c "from build import build_site; build_site()".)

  • Step 3: Remove the smoke-test manual and rebuild
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && rm -rf manuals-src/docs/test && venv/bin/python -c "from build import build_site; print(\"rebuilt:\", build_site()[0])"'

Expected: rebuilt: True.

Task 6: app.py — Flask server (static site + upload portal + job API)

Files:

  • Create: /home/levander/knowledgebase/app.py
  • Test: /home/levander/knowledgebase/test_app.py

Interfaces:

  • Consumes: slugs.safe_slug, ingest.submit, ingest.jobs, ingest.start_worker, ingest.DOCS

  • Produces: a Flask app; routes / + /<path:path> (static site), GET|POST /upload, GET /jobs/<id>, GET /api/jobs/<id>; main() starts the worker and serves via waitress on 127.0.0.1:8092

  • Step 1: Install flask + waitress

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/pip install -q flask waitress && venv/bin/python -c "import flask, waitress; print(\"flask\", flask.__version__)"'
  • Step 2: Write the failing test

Create /home/levander/knowledgebase/test_app.py:

import io
import unittest
 
from app import app
 
 
class TestUploadValidation(unittest.TestCase):
    def setUp(self):
        self.client = app.test_client()
 
    def test_get_upload_form(self):
        response = self.client.get("/upload")
        self.assertEqual(response.status_code, 200)
        self.assertIn(b"folder", response.data.lower())
 
    def test_rejects_non_pdf_extension(self):
        data = {"pdf": (io.BytesIO(b"hello"), "notes.txt"), "folder": "test", "name": "x"}
        response = self.client.post("/upload", data=data, content_type="multipart/form-data")
        self.assertEqual(response.status_code, 400)
 
    def test_rejects_bad_magic_bytes(self):
        data = {"pdf": (io.BytesIO(b"NOTPDF...."), "fake.pdf"), "folder": "test", "name": "x"}
        response = self.client.post("/upload", data=data, content_type="multipart/form-data")
        self.assertEqual(response.status_code, 400)
 
    def test_rejects_empty_folder(self):
        data = {"pdf": (io.BytesIO(b"%PDF-1.4 x"), "a.pdf"), "folder": "///", "name": "x"}
        response = self.client.post("/upload", data=data, content_type="multipart/form-data")
        self.assertEqual(response.status_code, 400)
 
    def test_api_unknown_job(self):
        response = self.client.get("/api/jobs/deadbeef")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.get_json()["state"], "unknown")
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 3: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest test_app -v'

Expected: FAIL with ModuleNotFoundError: No module named 'app'

  • Step 4: Write app.py

Create /home/levander/knowledgebase/app.py:

import os
 
from flask import Flask, jsonify, redirect, render_template_string, request, send_from_directory, url_for
from waitress import serve
 
from ingest import DOCS, jobs, start_worker, submit
 
PROJECT = os.path.dirname(os.path.abspath(__file__))
SITE = os.path.join(PROJECT, "site")
STAGE = os.path.join(PROJECT, "staging")
 
from slugs import safe_slug
 
app = Flask(__name__)
 
UPLOAD_HTML = """<!doctype html><html><head><meta name=viewport content="width=device-width,initial-scale=1">
<title>Add a manual</title><style>body{font-family:system-ui;max-width:40em;margin:2em auto;padding:0 1em}
input,button{font-size:1em;padding:.5em;margin:.3em 0;width:100%}.err{color:#b00}</style></head><body>
<h1>Add a manual</h1>{% if error %}<p class=err>{{error}}</p>{% endif %}
<form method=post enctype=multipart/form-data>
<label>PDF<input type=file name=pdf accept=application/pdf required></label>
<label>Folder<input name=folder list=folders required></label>
<datalist id=folders>{% for f in folders %}<option value="{{f}}">{% endfor %}</datalist>
<label>Manual name<input name=name required></label>
<button type=submit>Process</button></form>
<p><a href="/">Back to library</a></p></body></html>"""
 
JOB_HTML = """<!doctype html><html><head><meta name=viewport content="width=device-width,initial-scale=1">
<title>Processing</title><style>body{font-family:system-ui;max-width:40em;margin:2em auto;padding:0 1em}</style>
</head><body><h1>Processing</h1><p id=state>starting...</p><pre id=detail></pre>
<script>
async function poll(){
  const r = await fetch("/api/jobs/{{job_id}}"); const j = await r.json();
  document.getElementById("state").textContent = j.step || j.state;
  document.getElementById("detail").textContent = j.detail || "";
  if(j.state==="done"){document.getElementById("state").innerHTML="Done. <a href=/>Open library</a>";return;}
  if(j.state==="failed"){document.getElementById("state").textContent="Failed: "+(j.step||"");return;}
  setTimeout(poll,3000);
}
poll();
</script></body></html>"""
 
 
def _folders():
    if os.path.exists(DOCS):
        return sorted(entry for entry in os.listdir(DOCS) if os.path.isdir(os.path.join(DOCS, entry)))
    return []
 
 
@app.route("/upload", methods=["GET", "POST"])
def upload():
    if request.method == "GET":
        return render_template_string(UPLOAD_HTML, folders=_folders(), error=None)
    uploaded = request.files.get("pdf")
    if uploaded is None or not uploaded.filename.lower().endswith(".pdf"):
        return render_template_string(UPLOAD_HTML, folders=_folders(), error="A PDF file is required"), 400
    head = uploaded.stream.read(5)
    uploaded.stream.seek(0)
    if head[:4] != b"%PDF":
        return render_template_string(UPLOAD_HTML, folders=_folders(), error="Not a valid PDF"), 400
    try:
        folder = safe_slug(request.form.get("folder", ""))
        name = safe_slug(request.form.get("name", ""))
    except ValueError:
        return render_template_string(UPLOAD_HTML, folders=_folders(), error="Invalid folder or name"), 400
    os.makedirs(STAGE, exist_ok=True)
    staged = os.path.join(STAGE, name + ".pdf")
    uploaded.save(staged)
    title = request.form.get("name") or name
    job_id = submit(staged, folder, name, title)
    return redirect(url_for("job_page", job_id=job_id))
 
 
@app.route("/jobs/<job_id>")
def job_page(job_id):
    return render_template_string(JOB_HTML, job_id=job_id)
 
 
@app.route("/api/jobs/<job_id>")
def job_api(job_id):
    return jsonify(jobs.get(job_id, {"state": "unknown", "step": "unknown", "detail": ""}))
 
 
@app.route("/")
@app.route("/<path:path>")
def site(path=""):
    candidate = os.path.normpath(os.path.join(SITE, path))
    if not candidate.startswith(SITE):
        return "not found", 404
    if path == "" or os.path.isdir(candidate):
        path = os.path.join(path, "index.html")
    if not os.path.exists(os.path.join(SITE, path)):
        fallback = "404.html"
        if os.path.exists(os.path.join(SITE, fallback)):
            return send_from_directory(SITE, fallback), 404
        return "not found", 404
    return send_from_directory(SITE, path)
 
 
def main():
    start_worker()
    serve(app, host="127.0.0.1", port=8092)
 
 
if __name__ == "__main__":
    main()
  • Step 5: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest test_app -v'

Expected: OK — 5 tests pass.

  • Step 6: Full suite
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest discover -p "test_*.py" 2>&1 | tail -3'

Expected: OK — 25 tests pass (12 + 8 + 5).

Task 7: systemd service + tailscale serve + end-to-end verification

Files:

  • Create: /etc/systemd/system/knowledgebase.service

  • Step 1: Install and enable the service

/usr/bin/ssh levander@telep-mainframe 'sudo tee /etc/systemd/system/knowledgebase.service > /dev/null <<UNIT
[Unit]
Description=Knowledgebase — manual library + OCR ingest portal
After=network-online.target docker.service
Wants=network-online.target
 
[Service]
Type=simple
User=levander
WorkingDirectory=/home/levander/knowledgebase
ExecStart=/home/levander/knowledgebase/venv/bin/python /home/levander/knowledgebase/app.py
Restart=always
RestartSec=5
 
[Install]
WantedBy=multi-user.target
UNIT
sudo systemctl daemon-reload && sudo systemctl enable --now knowledgebase.service && sleep 4 && systemctl is-active knowledgebase.service'

Expected: active.

  • Step 2: Confirm it serves the library locally
/usr/bin/ssh levander@telep-mainframe 'curl -s -o /dev/null -w "root=%{http_code}\n" http://127.0.0.1:8092/; curl -s -o /dev/null -w "upload=%{http_code}\n" http://127.0.0.1:8092/upload; curl -s http://127.0.0.1:8092/ | grep -o "<title>[^<]*</title>" | head -1'

Expected: root=200, upload=200, and a title containing “Knowledgebase”.

  • Step 3: Stand up a dedicated knowledgebase tailnet node (second tailscaled on the host)

The service must be its own tailnet device named knowledgebase, not a port on telep-mainframe. Use a second tailscaled instance in userspace-networking mode.

First create the second tailscaled as a systemd unit and start it:

/usr/bin/ssh levander@telep-mainframe 'sudo mkdir -p /var/lib/tailscale-kb /run/tailscale-kb && sudo tee /etc/systemd/system/tailscaled-knowledgebase.service > /dev/null <<UNIT
[Unit]
Description=tailscaled (knowledgebase node)
After=network-online.target
Wants=network-online.target
 
[Service]
ExecStartPre=/bin/mkdir -p /run/tailscale-kb
ExecStart=/usr/sbin/tailscaled --tun=userspace-networking --state=/var/lib/tailscale-kb/tailscaled.state --socket=/run/tailscale-kb/tailscaled.sock --port=0
Restart=always
RestartSec=5
 
[Install]
WantedBy=multi-user.target
UNIT
sudo systemctl daemon-reload && sudo systemctl enable --now tailscaled-knowledgebase.service && sleep 3 && systemctl is-active tailscaled-knowledgebase.service'

Expected: active.

Then bring the node up — this needs ONE interactive login (hand the printed URL to the user, exactly like exit-vpn was authenticated). Run it detached so the auth URL stays live:

/usr/bin/ssh levander@telep-mainframe 'sudo nohup tailscale --socket=/run/tailscale-kb/tailscaled.sock up --hostname=knowledgebase --accept-dns=false > /tmp/kb-up.log 2>&1 & sleep 8; grep -A2 -i authenticate /tmp/kb-up.log | head -5'

Give the user the printed https://login.tailscale.com/a/... URL and WAIT for them to approve. Then confirm and add the serve mapping:

/usr/bin/ssh levander@telep-mainframe 'sudo tailscale --socket=/run/tailscale-kb/tailscaled.sock status 2>&1 | head -3; sudo tailscale --socket=/run/tailscale-kb/tailscaled.sock serve --bg --https 443 http://127.0.0.1:8092 2>&1 | tail -3; sudo tailscale --socket=/run/tailscale-kb/tailscaled.sock serve status 2>&1'

Expected: node knowledgebase logged in, and :443 mapped to 127.0.0.1:8092.

  • Step 4: End-to-end via the tailnet (from the Mac)
curl -s -o /dev/null -w "root=%{http_code}\n" https://knowledgebase.taild4189d.ts.net/
curl -s https://knowledgebase.taild4189d.ts.net/ | grep -o "Knowledgebase" | head -1
curl -s -o /dev/null -w "vitara page=%{http_code}\n" "https://knowledgebase.taild4189d.ts.net/suzuki-vitara/5door-supplement/001-introduction/"

Expected: root=200, Knowledgebase, and a 200 for the Vitara intro section page. (Bare folder URLs like .../5door-supplement/ return 404 by design — sections have no folder index; the nav links to section pages, not folders. The exact taild4189d tailnet suffix: confirm from tailscale status if the URL differs.)

  • Step 5: End-to-end portal ingest

Open https://knowledgebase.taild4189d.ts.net/upload from a tailnet device (or the Mac browser). Upload a small PDF, set folder test and name portal-check. Confirm the job page progresses to Done, then the manual appears in the nav and search. This is the true user-facing verification. Report the observed job states and whether the new manual rendered.

Afterwards remove it:

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && rm -rf manuals-src/docs/test && venv/bin/python -c "from build import build_site; print(build_site()[0])"'
  • Step 6: Phone / responsive check + lazy images

Load https://knowledgebase.taild4189d.ts.net/ on a phone. Confirm: responsive layout, folder nav works, search returns hits, and a manual page’s images load lazily (view-source shows loading="lazy" on <img>). Report what you observed.

curl -s "https://knowledgebase.taild4189d.ts.net/suzuki-vitara/5door-supplement/002-engine-mechanical/" 2>/dev/null | grep -c 'loading="lazy"' || true

Expected: a non-zero count of lazy images on a page that has figures.

  • Step 7: Confirm enabled for reboot
/usr/bin/ssh levander@telep-mainframe 'systemctl is-enabled knowledgebase.service'

Expected: enabled.

Self-review notes

Spec coverage: browsable folder tree (docs/ + mkdocs auto-nav, Task 3/4), mkdocs-material renderer with search + responsive + dark mode (Task 3), chunked loading via per-section split (Task 1) + lazy images (Task 3 hook, verified Task 7 Step 6), OCR ingest portal reusing ~/ocr venv (Task 5/6), user-picks-folder filing (Task 6 form), path-traversal guard (Task 2, enforced in Task 5/6, tested Task 7 Step 5), PDF-only magic-byte check (Task 6), atomic build swap so a bad ingest never breaks the live site (Task 3, verified Step 5), single service on the tailnet (Task 7), seed the two existing manuals (Task 4), phone verification (Task 7 Step 6).

Deliberate deviations from skill defaults: git commit steps omitted (user’s no-commit rule). The Task 5 smoke test OCRs a full manual (minutes) because there is no tiny sample PDF handy; it is the honest end-to-end worker test.

Graph View

Backlinks