KB Consolidated Generation (Phase 3) Implementation Plan

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

Goal: On-demand generate a consolidated, fully-cited how-to article from a cross-source cluster via claude -p --model opus, as a reviewable draft that publishes into the knowledgebase.

Architecture: Add generation modules to /home/levander/knowledgebase/ (co-located with the app so they share its venv and read the KB docs directly). A background worker gathers a cluster’s member sections, calls claude -p with a never-invent prompt, writes a draft + spec-check warnings; app routes render the draft and gate publish/discard.

Tech Stack: Python (knowledgebase venv: flask/waitress already there; stdlib for the rest), claude -p --model opus (installed at /home/levander/.local/bin/claude, logged in as levander), the knowledgebase Flask app, clusters.json from phase 2.

Design: 2026-07-25-kb-consolidated-gen-phase3-design

Global Constraints

All code under /home/levander/knowledgebase/ (co-located with app.py). Runs as levander, NOT root (claude is logged in as levander; app is levander). The sudo systemctl restart knowledgebase in the routes task is the only sudo.
No comments/docstrings/annotations in any code (hard user rule). Self-explanatory naming only.
Do not run git commit. Each task ends in verification.
Read-only over the source manuals and Qdrant. Generation only READS docs/<manual>/*.md + images and clusters.json; it WRITES only drafts/ and (on publish) docs/consolidated/.
Generator: claude -p --model opus (full path /home/levander/.local/bin/claude), prompt on stdin. —model opus is explicit — never Sonnet; fail loudly if it errors, never publish a failed/empty generation.
Single serialized background worker (mirror ingest.py’s queue/thread pattern) — one generation at a time (box is power-flaky + claude is heavy). Job states: queued → generating → done/failed.
clusters.json at /home/levander/kb-vectors/clusters.json; cluster member shape {manual_id, page_url, heading}; a member’s markdown = docs/<page_url rstrip "/">.md.
Publish target docs/consolidated/<slug>/ MUST be realpath-contained via slugs.resolve_within (reuse the existing traversal guard). Drafts + published articles carry an AI-provenance banner/header — never presented as a manual.
New app routes must be registered BEFORE the catch-all /<path:path> site route and must not regress existing routes/tests.
Run commands on the box: /usr/bin/ssh levander@telep-mainframe. On this Mac bare ssh is broken — always /usr/bin/ssh. Prefix any claude/heavy call with export PATH="$HOME/.local/bin:$PATH" or use the full path.

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

Task 1: genprompt.py — build the never-invent prompt

Files:

Create: /home/levander/knowledgebase/genprompt.py
Test: /home/levander/knowledgebase/test_genprompt.py

Interfaces:

Produces:
- member_md_path(page_url, docs_dir) -> str — docs_dir + "/" + page_url.rstrip("/") + ".md"
- build_prompt(cluster, sources) -> str — cluster = {label, members:[{manual_id,page_url,heading}]}; sources = list of (manual_id, heading, markdown); returns the full prompt string (rules + delimited labeled sources)
Step 1: Write the failing test

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

import unittest
 
from genprompt import build_prompt, member_md_path
 
 
class TestMemberPath(unittest.TestCase):
    def test_path(self):
        self.assertEqual(
            member_md_path("suzuki-vitara/workshop/007-diff/", "/docs"),
            "/docs/suzuki-vitara/workshop/007-diff.md",
        )
 
 
class TestBuildPrompt(unittest.TestCase):
    def _cluster(self):
        return {"label": "Differential", "members": [
            {"manual_id": "vitara/workshop", "page_url": "vitara/workshop/007-diff/", "heading": "Differential"},
            {"manual_id": "chevy/geo", "page_url": "chevy/geo/005-axle/", "heading": "Rear axle"},
        ]}
 
    def _sources(self):
        return [
            ("vitara/workshop", "Differential", "Torque to 40 Nm."),
            ("chevy/geo", "Rear axle", "Torque to 35 Nm."),
        ]
 
    def test_has_label(self):
        p = build_prompt(self._cluster(), self._sources())
        self.assertIn("Differential", p)
 
    def test_has_all_source_tags(self):
        p = build_prompt(self._cluster(), self._sources())
        self.assertIn("vitara/workshop", p)
        self.assertIn("chevy/geo", p)
 
    def test_has_source_markdown(self):
        p = build_prompt(self._cluster(), self._sources())
        self.assertIn("Torque to 40 Nm.", p)
        self.assertIn("Torque to 35 Nm.", p)
 
    def test_has_never_invent_rules(self):
        p = build_prompt(self._cluster(), self._sources())
        low = p.lower()
        self.assertIn("only", low)
        self.assertIn("verbatim", low)
        self.assertIn("[<manual_id>]".lower(), low) or self.assertIn("tag", low)
 
    def test_sources_delimited(self):
        p = build_prompt(self._cluster(), self._sources())
        self.assertEqual(p.count("=== SOURCE"), 2)
 
 
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_genprompt -v'

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

Step 3: Write the implementation

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

RULES = """You are consolidating multiple car-repair manual sections into ONE how-to article on: %s
 
HARD RULES:
1. Use ONLY facts present in the provided sources. Add nothing that is not in them.
2. Tag every specification, step, torque value, measurement, and part number with its source in the form [<manual_id>].
3. Reproduce all numbers, units, torque values, and wire colors VERBATIM. Never round, convert, or infer a value.
4. Where sources DISAGREE on a value or step, present BOTH with their tags and a warning marker. Do not pick one.
5. Carry through relevant source images: keep their markdown image references where the text references them, with the source tag.
6. If the sources do not cover something, omit it. Do not fill gaps.
 
OUTPUT: clean markdown. Start with "# %s", then a short intro naming the sources, then the consolidated procedure. No preamble or postamble outside the article.
 
SOURCES:
"""
 
 
def member_md_path(page_url, docs_dir):
    return docs_dir + "/" + page_url.rstrip("/") + ".md"
 
 
def build_prompt(cluster, sources):
    label = cluster["label"]
    parts = [RULES % (label, label)]
    for manual_id, heading, markdown in sources:
        parts.append("=== SOURCE [%s] — %s ===\n%s\n" % (manual_id, heading, markdown))
    return "\n".join(parts)

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_genprompt -v'

Expected: OK — 6 tests pass.

Task 2: speccheck.py — flag draft numbers absent from sources

Files:

Create: /home/levander/knowledgebase/speccheck.py
Test: /home/levander/knowledgebase/test_speccheck.py

Interfaces:

Produces:
- numbers(text) -> set[str] — number tokens (int/decimal, lowercased, whitespace-normalized), e.g. 40, 1.5, 35
- check(draft, sources_text) -> list[dict] — for each number in draft not present in sources_text, {number, context} (≤40 chars around the first occurrence)
Step 1: Write the failing test

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

import unittest
 
from speccheck import check, numbers
 
 
class TestNumbers(unittest.TestCase):
    def test_extracts(self):
        self.assertEqual(numbers("torque to 40 Nm and 1.5 mm"), {"40", "1.5"})
 
    def test_empty(self):
        self.assertEqual(numbers("no digits here"), set())
 
 
class TestCheck(unittest.TestCase):
    def test_flags_absent_number(self):
        result = check("Torque to 99 Nm.", "The manual says torque to 40 Nm.")
        self.assertEqual(len(result), 1)
        self.assertEqual(result[0]["number"], "99")
 
    def test_present_number_not_flagged(self):
        result = check("Torque to 40 Nm.", "torque to 40 Nm somewhere")
        self.assertEqual(result, [])
 
    def test_context_included(self):
        result = check("Set gap to 0.8 units.", "gap is 0.7")
        self.assertIn("0.8", result[0]["context"])
 
    def test_decimal_handling(self):
        result = check("Value 1.25 here", "value is 1.25 exactly")
        self.assertEqual(result, [])
 
 
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_speccheck -v'

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

Step 3: Write the implementation

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

import re
 
_NUM = re.compile(r"\d+(?:\.\d+)?")
 
 
def numbers(text):
    return set(_NUM.findall(text))
 
 
def check(draft, sources_text):
    source_numbers = numbers(sources_text)
    warnings = []
    seen = set()
    for match in _NUM.finditer(draft):
        token = match.group(0)
        if token in source_numbers or token in seen:
            continue
        seen.add(token)
        start = max(0, match.start() - 20)
        end = min(len(draft), match.end() + 20)
        warnings.append({"number": token, "context": draft[start:end].replace("\n", " ").strip()})
    return warnings

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_speccheck -v'

Expected: OK — 6 tests pass.

Task 3: gen.py (worker) + kbgen.py (CLI) — real generation

Files:

Create: /home/levander/knowledgebase/gen.py
Create: /home/levander/knowledgebase/kbgen.py

Interfaces:

Consumes: genprompt.build_prompt, genprompt.member_md_path, speccheck.check, slugs.safe_slug, build.build_site
Produces:
- gen.submit(cluster_id) -> str (job id), gen.jobs (dict), gen.start_worker(), gen.DRAFTS, gen.load_cluster(cluster_id) -> dict|None, gen.generate_draft(cluster) -> str (slug)
- gen.publish(slug) -> str (published path), gen.discard(slug), gen.list_drafts() -> list[dict]
Step 1: Write gen.py

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

import json
import os
import queue
import re
import shutil
import subprocess
import threading
import time
import uuid
 
from build import build_site
from genprompt import build_prompt, member_md_path
from slugs import resolve_within, safe_slug
from speccheck import check
 
PROJECT = os.path.dirname(os.path.abspath(__file__))
DOCS = os.path.join(PROJECT, "manuals-src", "docs")
CONSOLIDATED = os.path.join(DOCS, "consolidated")
DRAFTS = os.path.join(PROJECT, "drafts")
CLUSTERS_PATH = "/home/levander/kb-vectors/clusters.json"
CLAUDE = "/home/levander/.local/bin/claude"
IMG = re.compile(r"!\[[^\]]*\]\(([^)]+)\)")
 
jobs = {}
_queue = queue.Queue()
 
 
def load_cluster(cluster_id):
    try:
        with open(CLUSTERS_PATH, encoding="utf-8") as handle:
            for cluster in json.load(handle).get("clusters", []):
                if cluster["id"] == cluster_id:
                    return cluster
    except (OSError, ValueError):
        return None
    return None
 
 
def _gather_sources(cluster):
    sources = []
    source_text = []
    for member in cluster["members"]:
        path = member_md_path(member["page_url"], DOCS)
        if not os.path.exists(path):
            continue
        with open(path, encoding="utf-8") as handle:
            markdown = handle.read()
        sources.append((member["manual_id"], member["heading"], markdown))
        source_text.append(markdown)
    return sources, "\n".join(source_text)
 
 
def _copy_referenced_images(draft_md, cluster, dest_dir):
    referenced = set(IMG.findall(draft_md))
    for member in cluster["members"]:
        src_dir = os.path.dirname(member_md_path(member["page_url"], DOCS))
        for name in referenced:
            candidate = os.path.join(src_dir, os.path.basename(name))
            target = os.path.join(dest_dir, os.path.basename(name))
            if os.path.exists(candidate) and not os.path.exists(target):
                shutil.copy2(candidate, target)
 
 
def generate_draft(cluster):
    sources, source_text = _gather_sources(cluster)
    prompt = build_prompt(cluster, sources)
    result = subprocess.run(
        [CLAUDE, "-p", "--model", "opus"],
        input=prompt, capture_output=True, text=True, timeout=900,
    )
    if result.returncode != 0 or not result.stdout.strip():
        raise RuntimeError("claude generation failed: " + (result.stderr or "empty output")[-500:])
    draft_md = result.stdout.strip()
    slug = safe_slug(cluster["label"])
    dest = resolve_within(DRAFTS, slug)
    shutil.rmtree(dest, ignore_errors=True)
    os.makedirs(dest, exist_ok=True)
    with open(os.path.join(dest, "draft.md"), "w", encoding="utf-8") as handle:
        handle.write(draft_md)
    _copy_referenced_images(draft_md, cluster, dest)
    warnings = check(draft_md, source_text)
    meta = {
        "cluster_id": cluster["id"], "label": cluster["label"], "slug": slug,
        "sources": sorted(set(m["manual_id"] for m in cluster["members"])),
        "state": "draft", "speccheck": warnings,
    }
    with open(os.path.join(dest, "meta.json"), "w", encoding="utf-8") as handle:
        json.dump(meta, handle, ensure_ascii=False, indent=1)
    return slug
 
 
def submit(cluster_id):
    job_id = uuid.uuid4().hex[:12]
    jobs[job_id] = {"state": "queued", "step": "queued", "detail": ""}
    _queue.put((job_id, cluster_id))
    return job_id
 
 
def _process(job_id, cluster_id):
    try:
        cluster = load_cluster(cluster_id)
        if cluster is None:
            jobs[job_id].update(state="failed", step="no such cluster", detail=str(cluster_id))
            return
        jobs[job_id].update(state="generating", step="generating with claude (may take a while)")
        slug = generate_draft(cluster)
        jobs[job_id].update(state="done", step="done", detail=slug)
    except Exception as error:
        jobs[job_id].update(state="failed", step="error", detail=str(error)[-500:])
 
 
def _worker():
    while True:
        args = _queue.get()
        _process(*args)
        _queue.task_done()
 
 
def start_worker():
    threading.Thread(target=_worker, daemon=True).start()
 
 
def list_drafts():
    if not os.path.exists(DRAFTS):
        return []
    out = []
    for slug in sorted(os.listdir(DRAFTS)):
        meta_path = os.path.join(DRAFTS, slug, "meta.json")
        if os.path.exists(meta_path):
            with open(meta_path, encoding="utf-8") as handle:
                out.append(json.load(handle))
    return out
 
 
def read_draft(slug):
    safe = safe_slug(slug)
    base = resolve_within(DRAFTS, safe)
    md_path = os.path.join(base, "draft.md")
    meta_path = os.path.join(base, "meta.json")
    if not os.path.exists(md_path):
        return None
    with open(md_path, encoding="utf-8") as handle:
        markdown = handle.read()
    meta = {}
    if os.path.exists(meta_path):
        with open(meta_path, encoding="utf-8") as handle:
            meta = json.load(handle)
    return {"markdown": markdown, "meta": meta}
 
 
def publish(slug):
    safe = safe_slug(slug)
    src = resolve_within(DRAFTS, safe)
    if not os.path.exists(os.path.join(src, "draft.md")):
        raise RuntimeError("no such draft")
    dest = resolve_within(CONSOLIDATED, safe)
    shutil.rmtree(dest, ignore_errors=True)
    os.makedirs(dest, exist_ok=True)
    with open(os.path.join(src, "draft.md"), encoding="utf-8") as handle:
        body = handle.read()
    header = "> AI-generalt osszefoglalo. Ellenorizd a forras-kezikonyvek alapjan.\n\n"
    with open(os.path.join(dest, "001-" + safe + ".md"), "w", encoding="utf-8") as handle:
        handle.write(header + body)
    for name in os.listdir(src):
        if name.lower().endswith((".jpg", ".jpeg", ".png", ".gif")):
            shutil.copy2(os.path.join(src, name), dest)
    ok, error = build_site()
    if not ok:
        raise RuntimeError("build failed: " + error[:300])
    shutil.rmtree(src, ignore_errors=True)
    return dest
 
 
def discard(slug):
    safe = safe_slug(slug)
    shutil.rmtree(resolve_within(DRAFTS, safe), ignore_errors=True)

Step 2: Write kbgen.py

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

import argparse
 
import gen
 
 
def main():
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    c = sub.add_parser("cluster")
    c.add_argument("id", type=int)
    sub.add_parser("list-drafts")
    p = sub.add_parser("publish")
    p.add_argument("slug")
    d = sub.add_parser("discard")
    d.add_argument("slug")
    args = parser.parse_args()
    if args.command == "cluster":
        cluster = gen.load_cluster(args.id)
        if cluster is None:
            print("no such cluster", args.id)
            return
        print("generating:", cluster["label"], "(", len(cluster["members"]), "sources )")
        slug = gen.generate_draft(cluster)
        draft = gen.read_draft(slug)
        print("draft:", slug, "| speccheck warnings:", len(draft["meta"].get("speccheck", [])))
    elif args.command == "list-drafts":
        for meta in gen.list_drafts():
            print(meta["slug"], "|", meta["label"], "| warnings:", len(meta.get("speccheck", [])))
    elif args.command == "publish":
        print("published:", gen.publish(args.slug))
    elif args.command == "discard":
        gen.discard(args.slug)
        print("discarded", args.slug)
 
 
if __name__ == "__main__":
    main()

Step 3: Generate a real draft from a cross-source cluster (the real proof)

Pick a real cross-source cluster id (span ≥ 2) and generate. This calls claude -p opus — takes a minute or two; run thread-capped and serialized (nothing else heavy).

/usr/bin/ssh levander@telep-mainframe 'export PATH="$HOME/.local/bin:$PATH"; cd /home/levander/knowledgebase && ID=$(python3 -c "import json; cs=json.load(open(\"/home/levander/kb-vectors/clusters.json\"))[\"clusters\"]; x=[c for c in cs if c[\"span\"]>=2 and any(k in c[\"label\"].lower() for k in [\"brake\",\"cooling\",\"axle\",\"throttle\"])]; print((x or [c for c in cs if c[\"span\"]>=2])[0][\"id\"])"); echo "cluster id: $ID"; timeout 900 venv/bin/python kbgen.py cluster $ID'

Expected: generating: <label> ( N sources ) then draft: <slug> | speccheck warnings: <k>. A minute or two of claude time.

Step 4: Faithfulness spot-check (the real quality proof)

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && SLUG=$(venv/bin/python kbgen.py list-drafts | head -1 | cut -d"|" -f1 | tr -d " "); echo "=== draft: $SLUG ==="; sed -n "1,60p" drafts/$SLUG/draft.md; echo "=== speccheck ==="; python3 -c "import json; m=json.load(open(\"drafts/$SLUG/meta.json\")); print(\"sources:\", m[\"sources\"]); [print(\"  WARN\", w[\"number\"], \":\", w[\"context\"]) for w in m[\"speccheck\"]]"'

Expected: the draft is a # <label> how-to that cites source tags [manual_id], reproduces specs verbatim, and (if sources disagreed) shows both with a warning. Speccheck lists any draft numbers not found in sources. Read it: are the citations real and the specs traceable? Report honestly. (Draft images copied into drafts/<slug>/.)

Task 4: app.py routes — generate / drafts / publish / discard

Files:

Modify: /home/levander/knowledgebase/app.py
Test: /home/levander/knowledgebase/test_gen_routes.py

Interfaces:

Consumes: gen.submit, gen.jobs, gen.start_worker, gen.list_drafts, gen.read_draft, gen.publish, gen.discard
Produces: POST /generate/<int:cluster_id>, GET /api/genjobs/<job_id>, GET /drafts, GET /drafts/<slug>, POST /drafts/<slug>/publish, POST /drafts/<slug>/discard
Step 1: Write the failing test

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

import os
import shutil
import unittest
 
import app as kbapp
import gen
 
 
class TestGenRoutes(unittest.TestCase):
    def setUp(self):
        self.client = kbapp.app.test_client()
        self.slug = "zzz-testdraft"
        os.makedirs(os.path.join(gen.DRAFTS, self.slug), exist_ok=True)
        with open(os.path.join(gen.DRAFTS, self.slug, "draft.md"), "w") as handle:
            handle.write("# Test\nTorque 40 Nm [vitara/workshop]\n")
        import json
        with open(os.path.join(gen.DRAFTS, self.slug, "meta.json"), "w") as handle:
            json.dump({"cluster_id": 99, "label": "Test", "slug": self.slug,
                       "sources": ["vitara/workshop"], "state": "draft", "speccheck": []}, handle)
 
    def tearDown(self):
        shutil.rmtree(os.path.join(gen.DRAFTS, self.slug), ignore_errors=True)
 
    def test_drafts_list(self):
        response = self.client.get("/drafts")
        self.assertEqual(response.status_code, 200)
        self.assertIn(b"zzz-testdraft", response.data)
 
    def test_draft_page_renders_with_banner(self):
        response = self.client.get("/drafts/" + self.slug)
        self.assertEqual(response.status_code, 200)
        self.assertIn("AI".encode(), response.data)
        self.assertIn(b"Torque 40 Nm", response.data)
 
    def test_draft_missing_404(self):
        response = self.client.get("/drafts/nope-nope")
        self.assertEqual(response.status_code, 404)
 
    def test_genjob_unknown(self):
        response = self.client.get("/api/genjobs/deadbeef")
        self.assertEqual(response.get_json()["state"], "unknown")
 
    def test_discard(self):
        response = self.client.post("/drafts/" + self.slug + "/discard")
        self.assertIn(response.status_code, (200, 302))
        self.assertFalse(os.path.exists(os.path.join(gen.DRAFTS, self.slug, "draft.md")))
 
 
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_gen_routes -v'

Expected: FAIL (routes 404 / gen import error in app).

Step 3: Add to app.py

In /home/levander/knowledgebase/app.py: add import gen near the other imports; in main() add gen.start_worker() alongside the existing start_worker(). The draft page renders raw markdown inside a <pre> (dependency-free — no markdown library needed; the published article goes through the normal mkdocs render at publish time). Add the template + routes BEFORE the catch-all /<path:path> site route:

DRAFT_HTML = """<!doctype html><html><head><meta charset=utf-8>
<meta name=viewport content="width=device-width,initial-scale=1"><title>Draft: {{meta.label}}</title>
<style>body{font-family:system-ui;max-width:60em;margin:1.5em auto;padding:0 1em;background:#0f1115;color:#e6e8ee}
a{color:#4f8cff}.warn{background:#3a2a10;border:1px solid #7a5a20;border-radius:8px;padding:.6em 1em;margin:.6em 0}
.ban{background:#3a1520;border:1px solid #7a2030;border-radius:8px;padding:.7em 1em;font-weight:600}
pre{white-space:pre-wrap;background:#1a1d24;border:1px solid #2a2f3a;border-radius:10px;padding:1em}
form{display:inline}button{font-size:1em;padding:.5em 1em;margin:.3em .3em 0 0;border-radius:8px;border:0;cursor:pointer}
.pub{background:#2f6bff;color:#fff}.dis{background:#5a2030;color:#fff}</style></head><body>
<div class=ban>⚠ AI-generált összefoglaló — ellenőrizd a forrás-kézikönyvek alapján.</div>
<p>Források: {{meta.sources|join(' · ')}}</p>
{% if meta.speccheck %}<div class=warn><b>Ellenőrizendő számok</b> (nem található a forrásokban):
{% for w in meta.speccheck %}<div>⚠ {{w.number}} — <code>{{w.context}}</code></div>{% endfor %}</div>{% endif %}
<form method=post action="/drafts/{{meta.slug}}/publish"><button class=pub>Közzététel</button></form>
<form method=post action="/drafts/{{meta.slug}}/discard"><button class=dis>Elvetés</button></form>
<pre>{{markdown}}</pre><p><a href="/drafts">← Piszkozatok</a></p></body></html>"""
 
DRAFTS_LIST_HTML = """<!doctype html><html><head><meta charset=utf-8>
<meta name=viewport content="width=device-width,initial-scale=1"><title>Piszkozatok</title>
<style>body{font-family:system-ui;max-width:50em;margin:1.5em auto;padding:0 1em;background:#0f1115;color:#e6e8ee}a{color:#4f8cff}</style>
</head><body><h1>Piszkozatok</h1>{% for d in drafts %}<p><a href="/drafts/{{d.slug}}">{{d.label}}</a>
{% if d.speccheck %}<span style="color:#e0a030">({{d.speccheck|length}} ⚠)</span>{% endif %}</p>{% endfor %}
{% if not drafts %}<p>Nincs piszkozat.</p>{% endif %}<p><a href="/semantics">← Szemantika</a></p></body></html>"""
 
 
@app.route("/generate/<int:cluster_id>", methods=["POST"])
def generate_route(cluster_id):
    job_id = gen.submit(cluster_id)
    return redirect(url_for("job_page", job_id=job_id))
 
 
@app.route("/api/genjobs/<job_id>")
def genjob_api(job_id):
    return jsonify(gen.jobs.get(job_id, {"state": "unknown", "step": "unknown", "detail": ""}))
 
 
@app.route("/drafts")
def drafts_list():
    return render_template_string(DRAFTS_LIST_HTML, drafts=gen.list_drafts())
 
 
@app.route("/drafts/<slug>")
def draft_page(slug):
    draft = gen.read_draft(slug)
    if draft is None:
        return "not found", 404
    return render_template_string(DRAFT_HTML, markdown=draft["markdown"], meta=draft["meta"])
 
 
@app.route("/drafts/<slug>/publish", methods=["POST"])
def draft_publish(slug):
    try:
        gen.publish(slug)
    except Exception as error:
        return str(error), 500
    return redirect("/consolidated/" + gen.safe_slug(slug) + "/001-" + gen.safe_slug(slug) + "/")
 
 
@app.route("/drafts/<slug>/discard", methods=["POST"])
def draft_discard(slug):
    gen.discard(slug)
    return redirect(url_for("drafts_list"))

Note: gen.submit/jobs reuse the existing /jobs/<id> job page + job_page; the job page already polls /api/jobs/<id> — change the generate flow to poll /api/genjobs/<id> by pointing generate_route at a dedicated job page OR reuse job_page and add gen.jobs lookups to /api/jobs. SIMPLEST: make /api/jobs/<id> also check gen.jobs — modify the existing job_api to return jsonify(jobs.get(job_id) or gen.jobs.get(job_id) or {"state":"unknown",...}). Do that small merge so the existing job page works for gen jobs. (gen.safe_slug = re-export: add from slugs import safe_slug already imported in app; use safe_slug directly, not gen.safe_slug.)

Apply the safe_slug fix: in the publish redirect use safe_slug(slug) (already imported in app.py), not gen.safe_slug.

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_gen_routes -v'

Expected: OK — 5 tests pass.

Step 5: Full suite + restart + live

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && venv/bin/python -m unittest discover -p "test_*.py" 2>&1 | tail -3'
/usr/bin/ssh levander@telep-mainframe 'sudo systemctl restart knowledgebase && sleep 4 && systemctl is-active knowledgebase'
curl -s -o /dev/null -w "drafts=%{http_code}\n" --max-time 15 "https://knowledgebase.taild4189d.ts.net/drafts"
curl -s -o /dev/null -w "site=%{http_code} upload=" --max-time 15 "https://knowledgebase.taild4189d.ts.net/"
curl -s -o /dev/null -w "%{http_code}\n" --max-time 15 "https://knowledgebase.taild4189d.ts.net/upload"

Expected: all app tests pass (81+ plus new), service active, /drafts = 200, existing / and /upload still 200.

Task 5: /semantics “Generál” button + full e2e

Files:

Modify: /home/levander/knowledgebase/app.py (the SEMANTICS_HTML template only)
Step 1: Add a Generál button per cluster in SEMANTICS_HTML

In the SEMANTICS_HTML template (added in phase 2), inside the per-cluster <div class=c>, after the members loop, add a generate form. Edit the template block so each cluster shows:

<form method=post action="/generate/{{c.id}}" style="margin-top:.4em"><button style="font-size:.85em;padding:.3em .7em;border-radius:8px;border:0;background:#2f6bff;color:#fff;cursor:pointer">Generál összefoglalót</button></form>

Apply it:

/usr/bin/ssh levander@telep-mainframe 'python3 - <<PY
path="/home/levander/knowledgebase/app.py"
s=open(path,encoding="utf-8").read()
needle="{% endfor %}\n</div>{% endfor %}{% if not clusters %}"
btn="{% endfor %}\n<form method=post action=\"/generate/{{c.id}}\" style=\"margin-top:.4em\"><button style=\"font-size:.85em;padding:.3em .7em;border-radius:8px;border:0;background:#2f6bff;color:#fff;cursor:pointer\">Generál összefoglalót</button></form>\n</div>{% endfor %}{% if not clusters %}"
if "Generál összefoglalót" in s:
    print("button already present")
elif needle in s:
    s=s.replace(needle, btn, 1)
    open(path,"w",encoding="utf-8").write(s)
    print("button added")
else:
    print("ANCHOR NOT FOUND - inspect SEMANTICS_HTML members loop and add the form manually")
PY'

(If the anchor is not found because the phase-2 template differs, read the SEMANTICS_HTML block and insert the <form method=post action="/generate/{{c.id}}"> button inside each cluster div, after the members loop.)

Step 2: Restart + verify the button is live

/usr/bin/ssh levander@telep-mainframe 'sudo systemctl restart knowledgebase && sleep 4'
curl -s --max-time 15 "https://knowledgebase.taild4189d.ts.net/semantics" | grep -oE "/generate/[0-9]+" | head -3

Expected: several /generate/<id> form actions present (one per cluster).

Step 3: Full end-to-end — generate → draft → publish → live in KB

Drive the real flow over the tailnet: trigger generation for a cross-source cluster, poll the job, open the draft, publish, confirm it renders in the KB.

B="https://knowledgebase.taild4189d.ts.net"
ID=$(/usr/bin/ssh levander@telep-mainframe 'python3 -c "import json; cs=json.load(open(\"/home/levander/kb-vectors/clusters.json\"))[\"clusters\"]; print([c for c in cs if c[\"span\"]>=2][0][\"id\"])"')
echo "cluster: $ID"
JOB=$(curl -s -o /dev/null -w "%{redirect_url}" -X POST --max-time 30 "$B/generate/$ID")
JID=$(echo "$JOB" | grep -oE '[^/]+$')
echo "job: $JID"
for i in $(seq 1 40); do st=$(curl -s --max-time 15 "$B/api/genjobs/$JID" | python3 -c "import json,sys;print(json.load(sys.stdin).get('state'))" 2>/dev/null); echo "  $state $st"; [ "$st" = "done" ] && break; [ "$st" = "failed" ] && break; sleep 8; done
SLUG=$(curl -s --max-time 15 "$B/drafts" | grep -oE '/drafts/[a-z0-9-]+' | grep -v '/drafts$' | head -1 | sed 's#/drafts/##')
echo "draft slug: $SLUG"
curl -s -o /dev/null -w "draft-page=%{http_code}\n" --max-time 15 "$B/drafts/$SLUG"
curl -s -o /dev/null -w "publish=%{http_code}\n" -X POST --max-time 60 "$B/drafts/$SLUG/publish"
curl -s -o /dev/null -w "published-in-KB=%{http_code}\n" --max-time 15 "$B/consolidated/$SLUG/001-$SLUG/"

Expected: job reaches done, draft-page 200, publish 200/302, and the published article renders in the KB at /consolidated/<slug>/... = 200. Confirm the published page shows the AI-provenance header and cites sources.

Step 4: Confirm no source manuals were modified

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/knowledgebase && ls manuals-src/docs/ | grep -v consolidated | tr "\n" " "; echo; echo "consolidated: $(ls manuals-src/docs/consolidated/ 2>/dev/null)"'

Expected: the original manual folders unchanged; only a new consolidated/ folder added.

Self-review notes

Spec coverage: never-invent prompt contract with per-fact source tags + conflict-both-ways (Task 1); automated number-token spec-check flag (Task 2); background claude -p --model opus worker, on-demand one-at-a-time, draft written with images + meta (Task 3); draft render with AI banner + spec-check warnings, publish→docs/consolidated/ with build, discard (Tasks 3-4); /semantics Generál trigger (Task 5); traversal-guarded publish via resolve_within (Task 3); read-only over sources verified (Task 5 Step 4); real generation + faithfulness proof (Task 3 Steps 3-4) and full e2e (Task 5 Step 3); opus explicit, fail-loud (Task 3 generate_draft).

Deliberate deviations: modules co-located in knowledgebase/ not kb-vectors/ (avoids cross-venv import; they read KB docs + are driven by the KB app) — a refinement of the design’s placement. git commit steps omitted (user’s rule). The draft page renders raw markdown in a <pre> (dependency-free); the published article goes through the normal mkdocs render on build. /api/jobs is extended to also read gen.jobs so the existing job-page polling works for generation jobs.

Levandor

Explorer

KB consolidated generation (phase 3) — implementation plan

KB Consolidated Generation (Phase 3) Implementation Plan

Global Constraints

Task 1: genprompt.py — build the never-invent prompt

Task 2: speccheck.py — flag draft numbers absent from sources

Task 3: gen.py (worker) + kbgen.py (CLI) — real generation

Task 4: app.py routes — generate / drafts / publish / discard

Task 5: /semantics “Generál” button + full e2e

Self-review notes

Graph View

Table of Contents

Backlinks