"""The pages bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import * # noqa: F401,F403 — the shared wire helpers
# =============================================================================
# Bundle: pages — block presentation, diff URLs, computed display state,
# terminal rendering (rejected entries, ANSI SGR) and task-page layout.
# Fixture-driven: tasks and sessions are inserted as rows; blocks are
# raised through the push-era wire (POST /api/saddle/results with a
# blocked state) or inserted directly for terminal-task fixtures; only
# the rejected-ride scenario dispatches.
# =============================================================================
def run(ctx):
auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
app = ctx.app
def raise_block(session_id, question, kind="question", options=None, body=None):
"""The push-era block raise: a blocked round-results push
carrying the block payload (the poll-era /update channel is
gone — 404 below pins that)."""
block = {"question": question, "type": kind}
if options is not None:
block["options"] = options
if body is not None:
block["body"] = body
return saddle_api("POST", "/api/saddle/results", fake_token,
{"session_id": session_id, "state": "blocked",
"block": block})
check(True, "login yields an auth slot")
print("== the poll-era wire is gone (arch 19) ==")
# The daemon never called update/poll (verified against saddle/src);
# the routes 404 for any caller, authenticated or not.
status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
{"type": "status", "session_id": 1,
"data": {"state": "working"}})
check(status == 404, "POST /api/saddle/{id}/update is gone (404)")
status, _ = saddle_api("GET", "/api/saddle/%d/poll" % fake_id, fake_token)
check(status == 404, "GET /api/saddle/{id}/poll is gone (404)")
print("== blocks: push-era raises + body presentation ==")
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority,"
" created_at)"
" values (1, 'Block display task', 'working', 5,"
" strftime('%s','now'))")
plan_task = conn.execute("select max(id) from tasks").fetchone()[0]
conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
" values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
(plan_task, fake_id))
conn.commit()
session_id = conn.execute("select max(id) from sessions").fetchone()[0]
conn.close()
status, _ = raise_block(session_id, "Which database?", kind="question",
options=["SQLite", "Postgres"], body="# Storage choice")
check(status == 200, "question block raised through the results push")
print("== B1: highlight-to-comment quote dialog ==")
status, page = http("GET", "/tasks/%d" % plan_task, timeout=15)
check(status == 200 and "quote-comment-dialog" in page and "quote-text" in page,
"task page carries the quote-comment dialog the popover opens")
check('name="quote"' in page and 'name="return_to"' in page,
"quote dialog posts the quote and return-to fields")
status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
check(status == 200 and "quote-comment-dialog" in page,
"terminal page keeps its quote dialog")
print("== B3: review section hides the duplicate body ==")
status, _ = raise_block(session_id, "Review the diff", kind="review",
body="# The change\n\nReworks the flange module.")
check(status == 200, "review block raised through the results push")
detail = task_state(auth, plan_task)
review_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0]
check(detail.get("review_pending") is True,
"review section shows while the review is unanswered")
check(review_block.get("show_body") is False,
"pending review block hides its duplicate body preview")
check(review_block.get("body", "").startswith("# The change"),
"review body itself still arrives in state (B2 counterpart)")
act(auth, review_block["answer_action"], {"verdict": "accept"})
detail = task_state(auth, plan_task)
review_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0]
check(review_block.get("show_body") is True,
"answered review shows the body again (banner gone)")
check(review_block.get("answer") == "Review approved. Continue with the task.",
"the review accept form's verdict marker resolves to the stable wording")
# Plan 16 S2/R4: no verdict constant ships in any template — the
# accept forms carry only the verdict marker.
task_page_html = http_get_page(auth, "/tasks/%d" % plan_task)
check("Plan approved. Proceed with implementation." not in task_page_html
and "Review approved. Continue with the task." not in task_page_html,
"the task page ships no verdict constants (server-owned wording)")
check('name="verdict" value="accept"' in task_page_html,
"the accept forms post the verdict marker")
# Plan 16 S5 (U1): verdict-shaped blocks keep their explicit
# buttons + free-text path; a question with NULL options renders
# the question form (radio "other" + free reply), never a
# free-text-only verdict.
check("Request changes" in task_page_html and 'name="feedback"' in task_page_html,
"the review verdict keeps Request changes with free-text feedback (U1)")
check('value="__other__"' in task_page_html and 'name="other"' in task_page_html,
"question blocks render the option/other question form (U1)")
status, _ = raise_block(session_id, "Free-text thought?", kind="question")
check(status == 200, "a NULL-options question block raises")
bare_question = [b for b in task_state(auth, plan_task)["blocks"]
if b["question"] == "Free-text thought?"][0]
check(bare_question.get("options") == []
and bare_question.get("block_type") == "question",
"a NULL-options question stays a question (no options, not a verdict)")
print("== B2: question block body above options ==")
# Contract side (A2 will restore bodies end-to-end): a question
# block raised with a heredoc body carries it; the body preview
# rides above the options (template order) and survives answering
# alongside the selected answer. (Blocks list newest-first; the
# U1 fixture above raised a second, options-less question.)
detail = task_state(auth, plan_task)
question = [b for b in detail["blocks"]
if b["block_type"] == "question"
and b["body"] == "# Storage choice"][0]
check(question.get("body") == "# Storage choice"
and question.get("show_body") is True,
"question block keeps its body markdown with the preview on")
act(auth, question["answer_action"], {"option": "SQLite"})
detail = task_state(auth, plan_task)
question = [b for b in detail["blocks"]
if b["block_type"] == "question"
and b["body"] == "# Storage choice"][0]
check(question.get("answered") is True and question.get("answer") == "SQLite"
and question.get("body") == "# Storage choice"
and question.get("show_body") is True,
"answered question shows body + selected answer")
print("== B5: compare-changes SSH -> HTTPS ==")
def diff_url_for_project(git_url, suffix):
state = slot_public(auth, "/projects", "projects")
act(auth, state["create"], {"name": "diff-%s" % suffix, "git_url": git_url})
project_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
if p["name"] == "diff-%s" % suffix][0])
conn = db()
conn.execute("insert into tasks (project_id, title, status, task_branch,"
" branch_from, created_at) values (?, ?, 'working', 'feature/x',"
" 'main', strftime('%s','now'))", (project_id, "Diff %s" % suffix))
conn.commit()
task_id = conn.execute("select max(id) from tasks").fetchone()[0]
conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at,"
" outcome) values (?, ?, strftime('%s','now'), strftime('%s','now'),"
" 'done')", (task_id, fake_id))
conn.commit()
session = conn.execute("select max(id) from sessions").fetchone()[0]
conn.close()
status, _ = raise_block(session, "Review the diff", kind="review")
return task_state(auth, task_id).get("review_diff_url") or ""
check(diff_url_for_project("ssh://fabrica.unitatem.net:222/clanker/clanker-saddle.git",
"fabrica")
== "https://fabrica.unitatem.net/clanker/clanker-saddle/compare/main...feature/x",
"ssh://git@host:port/owner/repo.git translates (port dropped)")
check(diff_url_for_project("git@github.com:acme/widgets.git", "scp")
== "https://github.com/acme/widgets/compare/main...feature/x",
"git@host:owner/repo.git translates to the browse host")
check(diff_url_for_project("https://example.com/plain/repo", "plain")
== "https://example.com/plain/repo/compare/main...feature/x",
"HTTPS URLs pass through unchanged")
print("== B6: computed display state ==")
# Draft tasks display as draft (plan 11) and never dispatch.
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
"project_id": "1",
"title": "Draft display task",
"priority": "5",
})
draft_display = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
if t["title"] == "Draft display task"][0])
detail = task_state(auth, draft_display)
check(detail["status"] == "draft" and detail.get("display") == "draft",
"new tasks default to draft and display it")
check(detail.get("draft") is True and detail.get("enqueue") is not None
and detail.get("delete") is not None,
"draft tasks expose the enqueue + delete actions")
check(quiet_window(quiet_seconds(),
lambda: task_state(auth, draft_display)["status"] == "draft",
"draft task never dispatches"),
"a draft task is never dispatched or woken for")
# Plan 12: the waiting-on-drafts block is deleted (goal tasks are
# the goal agent's to sequence); a parked task displays plainly.
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
" values (1, 'Parked plain task', 'suspended', 5, strftime('%s','now'))")
conn.commit()
conn.close()
parked_id = db().execute("select max(id) from tasks").fetchone()[0]
detail = task_state(auth, parked_id)
check(detail.get("display") == "parked",
"a suspended task displays as parked (%s)" % detail.get("display"))
dash = slot_public(auth, "/", "dashboard")
check(not dash.get("waiting_drafts"),
"the waiting-on-drafts block is gone from the dashboard data")
status, home_page = http("GET", "/", timeout=15)
check(status == 200 and "Waiting on subtasks to be enqueued" not in home_page,
"the dashboard renders no waiting-on-drafts section")
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
" values (1, 'Parent', 'done', 5, strftime('%s','now'))")
parent6 = conn.execute("select max(id) from tasks").fetchone()[0]
# The post-mortem case: stored done WITH an unanswered review block.
# (A terminal task consumes result pushes silently — plan 13 B2 —
# so the row is inserted directly, the goalui deep-link pattern.)
conn.execute("insert into task_blocks (task_id, question, type, created_at)"
" values (?, 'Review the final diff', 'review',"
" strftime('%s','now'))", (parent6,))
conn.commit()
conn.close()
detail = task_state(auth, parent6)
check(detail.get("display") == "review requested" and detail.get("terminal") is True,
"unanswered review overrides the stored done status (terminal flag keeps stored truth)")
queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
if int(t["id"]) == parent6][0]
check(queue["display"] == "review requested",
"queue row shows the computed review-requested display")
# With the review answered the display falls back to the stored
# status (the plan-12 shape: no subtask-delegation states).
conn = db()
conn.execute("update task_blocks set answered_at = strftime('%s','now'),"
" answer = 'Human approved the review' where task_id = ?", (parent6,))
conn.commit()
conn.close()
detail = task_state(auth, parent6)
check(detail.get("display") == "done" and detail.get("terminal") is True,
"answered review falls back to the stored done display (%s)"
% detail.get("display"))
# Flags track the stored status on failure.
conn = db()
conn.execute("update tasks set status = 'failed', completed_at = strftime('%s','now')"
" where id = ?", (parent6,))
conn.commit()
conn.close()
detail = task_state(auth, parent6)
check(detail.get("display") == "failed" and detail.get("failed") is True
and detail.get("terminal") is True,
"failed task with answered review shows failed + terminal flags")
print("== rejected entries (non-fatal) ==")
# The push wire: a rejected entry rides a round-results push, stores
# on the session and streams through SSE; the task is untouched.
status, _ = saddle_api("POST", "/api/saddle/results", fake_token, {
"session_id": session_id, "state": "working",
"entries": [{
"id": 9021, "command": "bash -n script.sh",
"output": "syntax error near line 3 (recoverable)",
"exit_code": 1, "type": "rejected",
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}]})
check(status == 200, "a rejected entry rides the results push")
request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
with urllib.request.urlopen(request, timeout=15) as stream:
rejected_event = None
deadline = time.monotonic() + 8
while time.monotonic() < deadline and rejected_event is None:
line = stream.readline().decode().strip()
if not line.startswith("data: ") or line == "data: ping":
continue
event = json.loads(line[6:])
if event.get("type") == "rejected":
rejected_event = event
check(rejected_event is not None
and "syntax error near line 3" in (rejected_event.get("output") or ""),
"terminal stream carries rejected entries with their reason")
# The page template renders them with the distinct non-fatal style.
status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
check(status == 200 and "term-rejected" in page,
"terminal page renders rejected entries in their own style")
print("== B8: ANSI SGR in the terminal view ==")
# Byte-exact SGR delivery: the daemon's coloured chunks flow through
# the store + SSE untouched (the client renderer styles them).
coloured = "\x1b[1;32mbold green\x1b[0m then \x1b[38;5;196m256-red\x1b[0m\n"
conn = db()
conn.execute("insert into session_raw_log (session_id, seq, chunk)"
" values (?, 900, ?)", (session_id, coloured))
conn.commit()
conn.close()
request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
with urllib.request.urlopen(request, timeout=15) as stream:
coloured_event = None
deadline = time.monotonic() + 8
while time.monotonic() < deadline and coloured_event is None:
line = stream.readline().decode().strip()
if not line.startswith("data: ") or line == "data: ping":
continue
event = json.loads(line[6:])
if event.get("type") == "raw" and "\x1b[1;32m" in (event.get("text") or ""):
coloured_event = event
check(coloured_event is not None
and "\x1b[38;5;196m" in coloured_event["text"],
"SGR sequences stream byte-exact to the client renderer")
status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
check(status == 200 and "makeAnsiRenderer" in page and "colour256" in page,
"terminal page embeds the ANSI renderer (SGR + 256-colour palette)")
# Dispatch path: a rejected entry in a round's results push flows
# into the audit trail without failing (or finishing) the task.
task_id_tmp = create_queued(auth, {
"project_id": "1",
"title": "Rejected ride",
"priority": "2",
})
rejected_task = task_id_tmp
check(wait_for(lambda: task_state(auth, rejected_task).get("status") == "working"
or False, timeout=60, what="rejected-ride working") is not None,
"rejected-ride task reaches working")
fake.queue_output("rejected", "echo 'too long' | saddle",
"max_tokens exceeded (recoverable)")
check(wait_task_done(auth, rejected_task, timeout=60) is not None,
"task completes normally past rejected entries")
conn = db()
stored = conn.execute(
"select count(*) from session_commands sc join sessions s on s.id = sc.session_id"
" where s.task_id = ? and sc.type = 'rejected'", (rejected_task,)).fetchone()[0]
conn.close()
check(stored >= 1,
"rejected entries stored in the audit trail (%d rows)" % stored)
check(task_state(auth, rejected_task)["status"] == "done"
and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""),
"rejected entries never fail the task")
print("== plan 11 task D: task page — progress first, 50/50 grid, expanders ==")
detail = task_state(auth, plan_task)
block_ids = [int(b["id"]) for b in detail.get("blocks", [])]
check(len(block_ids) >= 3 and block_ids == sorted(block_ids, reverse=True),
"the blocks section orders newest-first via the sealed GET state (%s)"
% block_ids)
status, page = http("GET", "/tasks/%d" % plan_task, timeout=15)
# Plan 11 task D: the progress feed sits ABOVE the status card;
# status + description share a 50/50 grid; the plan detail hides
# behind its expander below it.
check(status == 200 and "grid-2" in page
and page.index("Progress
") < page.index("status-head"),
"the progress feed renders above the status card")
check(page.index("grid-2") < page.index("status-head")
and page.index("status-head") < page.index('data-long-form data-field="task"'),
"status and description sit side by side inside the grid (status first)")
check("Mandate" in page and "plan-expander" not in page,
"the task page carries no plan expander (plans live on the goal)")
# Answered blocks fold behind the history expander (this fixture's
# three blocks are all answered by now).
check("blocks-history" in page and "Answered blocks" in page,
"answered blocks hide behind the blocks history expander")
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority,"
" created_at) values (1, 'Progress layout task', 'working', 5,"
" strftime('%s','now','-120 seconds'))")
layout_task = conn.execute("select max(id) from tasks").fetchone()[0]
# backdated so the dashboard-timeline fixtures below stay the
# newest events
conn.execute("insert into task_progress (task_id, summary, created_at)"
" values (?, 'first update', strftime('%s','now','-90 seconds'))",
(layout_task,))
conn.execute("insert into task_progress (task_id, summary, created_at)"
" values (?, 'second update', strftime('%s','now','-89 seconds'))",
(layout_task,))
conn.commit()
conn.close()
status, history_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
check(status == 200
and history_page.index("progress-latest") < history_page.index("progress-history"),
"the progress timeline renders newest-first (latest above the history)")
# The edit expander sits UNDER the "Show history" expander.
check(history_page.index("Show history") < history_page.index("edit-expander"),
"the edit expander sits under the Show history expander")
print("== heredoc completion reports: markdown card + timeline snippet ==")
# The daemon's `cat << 'EOF' | saddle done` stores a markdown
# summary (the failure path stores the same field); the task page
# renders it as a full-width completion-report card directly below
# the status/description grid — markdown via stm-markdown
# (XSS-safe), "Completion report" when done, "Failure summary"
# when failed, no card when the summary is empty. The dashboard
# timeline snippets the summary's FIRST line.
report_md = ("## Flange wired and tested\n\n"
"Ran `meson test` over the full matrix.\n\n"
"```sh\nmeson test -C builddir\n```\n\n"
"not html & escaped\n")
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority,"
" summary, started_at, completed_at, created_at)"
" values (1, 'Report done task', 'done', 5, ?,"
" strftime('%s','now'), strftime('%s','now'),"
" strftime('%s','now'))", (report_md,))
report_done = conn.execute("select max(id) from tasks").fetchone()[0]
conn.execute("insert into tasks (project_id, title, status, priority,"
" summary, started_at, completed_at, created_at)"
" values (1, 'Report failed task', 'failed', 5,"
" 'Boom: tests failed\n\nsecond line carries the detail',"
" strftime('%s','now'), strftime('%s','now'),"
" strftime('%s','now'))")
report_failed = conn.execute("select max(id) from tasks").fetchone()[0]
conn.execute("insert into tasks (project_id, title, status, priority,"
" started_at, completed_at, created_at)"
" values (1, 'Report empty task', 'done', 5,"
" strftime('%s','now'), strftime('%s','now'),"
" strftime('%s','now'))")
report_empty = conn.execute("select max(id) from tasks").fetchone()[0]
conn.commit()
conn.close()
done_state = task_state(auth, report_done)
check(done_state["report_title"] == "Completion report"
and "## Flange wired" in (done_state.get("summary") or ""),
"a done task's summary rides the report card state")
check(task_state(auth, report_failed)["report_title"] == "Failure summary",
"a failed task's card heading reads Failure summary")
empty_state = task_state(auth, report_empty)
check(empty_state["report_title"] == "" and not (empty_state.get("summary") or ""),
"an empty summary renders no report card (no heading)")
check(task_state(auth, layout_task)["report_title"] == "",
"non-terminal tasks never show the report card")
status, page = http("GET", "/tasks/%d" % report_done, timeout=15)
check(status == 200 and "completion-report" in page
and page.index("grid-2") < page.index("completion-report")
< page.index("edit-expander"),
"the completion-report card sits below the status/description grid, full width")
card = page[page.index("completion-report"):page.index("edit-expander")]
check('stm-markdown="task.summary"' in card
and 'stm-text="task.report_title"' in card,
"the card renders the summary as markdown under its state-driven heading")
check("Summary:" not in page,
"the old plain-text summary line is gone (the card owns the summary)")
# The markdown itself renders client-side: drive the SHIPPED
# stm-markdown renderer (statum.js) on the fixture summary via
# node — headings and code come out as HTML, no markdown syntax
# leaks, raw HTML stays escaped.
status, statum_js = http("GET", "/_statum/resource/statum.js", timeout=15)
skip = None if status == 200 else "statum.js not served"
parts = []
if skip is None:
def extract_md(name, source):
start = source.index("function %s(" % name)
depth = 0
index = source.index("{", start)
end = index
while True:
if source[end] == "{":
depth += 1
elif source[end] == "}":
depth -= 1
if depth == 0:
break
end += 1
return source[start:end + 1]
try:
parts = [extract_md(name, statum_js)
for name in ("safeMarkdownUrl", "mdLinks", "mdInline", "renderMarkdown")]
except ValueError:
skip = "renderer functions missing"
rendered = ""
if skip is None:
try:
result = subprocess.run(
["node", "-e", "%s\n%s\n%s\n%s\nconsole.log(renderMarkdown(process.argv[1]));"
% tuple(parts), report_md],
capture_output=True, text=True, timeout=20)
rendered = result.stdout
except (OSError, subprocess.SubprocessError):
skip = "node unavailable"
if skip is not None:
check(True, "stm-markdown renderer present (%s, deep check skipped)" % skip)
else:
check("Flange wired and tested
" in rendered
and "meson test" in rendered,
"the report summary renders headings and code through stm-markdown")
check("## Flange" not in rendered and "```sh" not in rendered
and "`meson test`" not in rendered,
"no raw markdown syntax leaks into the rendered report")
check("not html" not in rendered and "<b>not html</b>" in rendered
and "& escaped" in rendered,
"raw HTML in the summary stays escaped (XSS-safe)")
# The dashboard timeline: completion/failure events snippet the
# summary's first line, falling back to the bare verdict text.
dash = slot_public(auth, "/", "dashboard")
done_event = next((e for e in dash.get("activity", [])
if e["kind"] == "completed" and int(e["task_id"]) == report_done), None)
check(done_event is not None
and done_event["text"] == "Task completed: ## Flange wired and tested",
"the timeline completion event snippets the summary's first line")
failed_event = next((e for e in dash.get("activity", [])
if e["kind"] == "failed" and int(e["task_id"]) == report_failed), None)
check(failed_event is not None
and failed_event["text"] == "Task failed: Boom: tests failed",
"the timeline failure event snippets the summary's first line")
empty_event = next((e for e in dash.get("activity", [])
if e["kind"] == "completed" and int(e["task_id"]) == report_empty), None)
check(empty_event is not None and empty_event["text"] == "Task completed",
"a missing summary falls back to the bare timeline text")
print("== plan 11 task D: block split + queue-screen layout ==")
# One open and one answered question: the open block renders in
# full, the answered one folds behind the history expander.
conn = db()
conn.execute("insert into sessions (task_id, saddle_id, started_at)"
" values (?, ?, strftime('%s','now'))", (layout_task, fake_id))
conn.commit()
split_session = conn.execute("select max(id) from sessions").fetchone()[0]
conn.close()
raise_block(split_session, "Open question for layout?", kind="question")
raise_block(split_session, "Answered question for layout?", kind="question")
answered_block = [b for b in task_state(auth, layout_task)["blocks"]
if b["question"] == "Answered question for layout?"][0]
act(auth, answered_block["answer_action"], {"answer": "Yes"})
detail = task_state(auth, layout_task)
check(len(detail.get("open_blocks", [])) == 1
and len(detail.get("answered_blocks", [])) == 1
and detail["open_blocks"][0]["question"] == "Open question for layout?"
and detail["answered_blocks"][0]["question"] == "Answered question for layout?",
"the task slot splits blocks into open (current) and answered lists")
status, page = http("GET", "/tasks/%d" % layout_task, timeout=15)
check(status == 200
and page.index('stm-for-block-in="task.open_blocks"') < page.index("blocks-history")
< page.index('stm-for-block-in="task.answered_blocks"'),
"current blocks render in full above the answered-behind-expander history")
check(page.index("status-head") < page.index("Cancel task"),
"the lifecycle buttons ride the status card inside the grid")
status, queue_page = http("GET", "/tasks", timeout=15)
check(status == 200
and queue_page.index("filter-status") < queue_page.index("` arrives as a round-results entry with the
# message in the output field: it stores like any entry type, earns
# the task its 🤠 badge (derived, nothing new stored), renders as the
# task page's festive card (last 3, newest first, "+N more") and
# surfaces on the dashboard timeline as its own event kind.
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
" values (1, 'Yeehaw task', 'working', 5, strftime('%s','now'))")
yeehaw_task = conn.execute("select max(id) from tasks").fetchone()[0]
conn.execute("insert into sessions (task_id, saddle_id, started_at)"
" values (?, ?, strftime('%s','now'))", (yeehaw_task, fake_id))
yeehaw_session = conn.execute("select max(id) from sessions").fetchone()[0]
conn.commit()
conn.close()
now_epoch = int(time.time())
def push_yeehaw(entry_id, message, offset=0):
"""One celebration riding a round-results push, timestamped
`offset` seconds from now (distinct epochs keep the timeline's
newest-first ordering observable)."""
return saddle_api("POST", "/api/saddle/results", fake_token, {
"session_id": yeehaw_session, "state": "working",
"entries": [{
"id": entry_id, "command": "saddle yeehaw", "output": message,
"exit_code": 0, "type": "yeehaw",
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ",
time.gmtime(now_epoch + offset)),
}]})[0]
def yeehaw_rows():
conn = db()
try:
return conn.execute(
"select count(*) from session_commands where session_id = ?"
" and type = 'yeehaw'", (yeehaw_session,)).fetchone()[0]
finally:
conn.close()
check(push_yeehaw(9101, "Shipped the flange") == 200,
"yeehaw entries ingest through the results push like any entry type")
check(yeehaw_rows() == 1,
"the yeehaw entry stores as a session_commands row (message in output)")
check(push_yeehaw(9101, "Shipped the flange") == 200 and yeehaw_rows() == 1,
"a redelivered yeehaw push is idempotent on the daemon entry id")
check(task_state(auth, yeehaw_task)["status"] == "working",
"a yeehaw entry never derails the task state")
push_yeehaw(9102, "Second hoedown", 1)
push_yeehaw(9103, "Third hoedown\nsecond line carries the detail", 2)
push_yeehaw(9104, "Fourth hoedown", 3)
push_yeehaw(9105, "Fifth hoedown", 4)
detail = task_state(auth, yeehaw_task)
check(detail.get("yeehaw") is True and detail.get("yeehaw_total") == 5,
"the detail state flags the badge presence and the uncapped total")
check([y["message"] for y in detail.get("yeehaws", [])]
== ["Fifth hoedown", "Fourth hoedown",
"Third hoedown\nsecond line carries the detail"],
"the festive card keeps the last 3 yeehaws, newest first (full messages)")
check(detail.get("yeehaw_more") == 2,
"the '+N more' hint counts the capped tail")
status, page = http("GET", "/tasks/%d" % yeehaw_task, timeout=15)
# The template compiler HTML-escapes the emoji (🤠 = 🤠).
check(status == 200 and "yeehaw-card" in page and "🤠 Yeehaw!" in page,
"the task page renders the festive yeehaw card")
check(page.index("Progress
") < page.index("yeehaw-card")
< page.index("status-head"),
"the card sits between the progress feed and the status grid")
check('stm-text="task.yeehaw_more"' in page and "more yeehaw" in page,
"the card carries the +N more hint")
check("badge-yeehaw" in page and "status-corner" in page,
"the status card corner carries the earned 🤠 badge")
plain_detail = task_state(auth, layout_task)
check(plain_detail.get("yeehaw") is False and plain_detail.get("yeehaw_total") == 0
and plain_detail.get("yeehaws", []) == [],
"tasks without yeehaw entries carry no card state")
status, plain_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
check(status == 200 and 'section class="card yeehaw-card" stm-if="task.yeehaw"' in plain_page,
"the card template only renders under state gating (stm-if task.yeehaw)")
queue_rows = slot_public(auth, "/tasks", "tasks")["tasks"]
check([t for t in queue_rows if int(t["id"]) == yeehaw_task][0].get("has_yeehaw") is True,
"queue rows badge yeehaw tasks")
check([t for t in queue_rows if int(t["id"]) == layout_task][0].get("has_yeehaw") is False,
"queue rows leave yeehaw-less tasks bare")
status, queue_page = http("GET", "/tasks", timeout=15)
check(status == 200 and 'stm-if="task.has_yeehaw"' in queue_page,
"the queue template renders the row badge")
dash = slot_public(auth, "/", "dashboard")
yeehaw_events = [e for e in dash.get("activity", []) if e["kind"] == "yeehaw"]
check(len(yeehaw_events) == 5
and all(int(e["task_id"]) == yeehaw_task for e in yeehaw_events),
"every yeehaw entry surfaces as its own timeline event kind")
check(yeehaw_events[0]["text"] == "🤖 yeehaw'd: Fifth hoedown",
"the newest yeehaw event leads the timeline rows")
check(all("second line" not in e["text"] for e in yeehaw_events),
"multiline yeehaw messages snippet to their first line")
referenced = [e for e in dash.get("activity", []) if int(e["task_id"]) == yeehaw_task]
check(len(referenced) > 0 and all(e.get("task_yeehaw") is True for e in referenced),
"dashboard timeline task references carry the badge flag")
plain_refs = [e for e in dash.get("activity", []) if int(e["task_id"]) == layout_task]
check(len(plain_refs) > 0 and all(e.get("task_yeehaw") is False for e in plain_refs),
"references to yeehaw-less tasks stay bare")
status, home_page = http("GET", "/", timeout=15)
check(status == 200 and 'stm-if="item.task_yeehaw"' in home_page
and "badge-yeehaw" in home_page,
"the dashboard renders the badge on timeline task references")
# Close the yeehaw fixture's round (a plain done push under its
# session): under the derived view (item 23 — busy iff an open
# session on an assigned/working task) the fixture round would
# hold the fake's saddle busy forever; the goal section below
# needs it free.
status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
{"session_id": yeehaw_session, "state": "done",
"summary": "yeehaw fixture round closed"})
check(wait_task_done(auth, yeehaw_task, timeout=30) is not None,
"the yeehaw fixture round closes (the derived busy view frees the fake)")
# (The old R10 scenario — branch_advanced riding /poll verdicts —
# died with the poll channel; the answers bundle pins the hint on
# its surviving surface, the resumed assignment payload.)
print("== plan 12: the web surfaces — pile, timeline, queue, task page ==")
# A goal parked in plan review feeds the unified dashboard pile;
# goal events feed the timeline.
act(auth, slot_public(auth, "/goals", "goals")["create"], {
"project_id": "1", "title": "Dashboard goal",
"brief": "Feeds the pile.", "branch_name": "goal/dash"})
conn = db()
goal3 = conn.execute(
"select id from goals where title = 'Dashboard goal'").fetchone()[0]
conn.close()
fake.goal_script = [("plan", "# Dash plan\n\n1. Probe the pile")]
act(auth, goal_state(goal3)["start"], {})
check(wait_for(lambda: goal_state(goal3)["status"] == "plan_review" or False,
timeout=60, what="dashboard goal enters plan review") is not None,
"the dashboard goal reaches plan review")
check(wait_for(lambda: any(b.get("item_type") == "goal"
and int(b.get("goal_id") or 0) == goal3
for b in slot_public(auth, "/", "dashboard")["pinned"])
or False, timeout=30, what="goal block pins on the dashboard") is not None,
"the unified dashboard pile carries the goal's plan review")
pile = slot_public(auth, "/", "dashboard")["pinned"]
check(all(b.get("item_type") in ("task", "goal") for b in pile),
"pile items carry type badges (task/goal interleaved)")
goal_pile = [b for b in pile if b.get("item_type") == "goal"][0]
check(goal_pile["block_type"] == "plan" and "Dashboard goal" in goal_pile["goal_title"],
"the goal pile card names the goal and block kind")
# Answer it: the pile drains.
act(auth, goal_state(goal3)["plan_pending_block"]["answer_action"],
{"answer": "Plan approved. Proceed with implementation."})
check(wait_for(lambda: not any(b.get("item_type") == "goal"
and int(b.get("goal_id") or 0) == goal3
for b in slot_public(auth, "/", "dashboard")["pinned"])
or False, timeout=30, what="goal pile card drains") is not None,
"answering the review drains the goal's pile card")
activity = slot_public(auth, "/", "dashboard")["activity"]
check(any(a["kind"] == "goal-plan-submitted" and int(a.get("goal_id") or 0) == goal3
and a["goal_title"] == "Dashboard goal" for a in activity),
"the dashboard timeline carries goal events linked to the goal")
# The task queue: goal filter options, mode badges, goal links,
# enqueue-after dropdown.
queue = slot_public(auth, "/tasks", "tasks")
check(any(int(g["id"]) == goal3 and g["title"] == "Dashboard goal"
for g in queue["goal_options"]),
"the queue's goal filter lists the goals")
check(queue["tasks"][0].get("mode") in ("implement", "investigate", "review", "test",
"orchestrate"),
"queue rows carry the plan-12 mode")
check(all(("goal_title" in t) and ("goal_id" in t) and ("creator" in t)
for t in queue["tasks"][:3]),
"queue rows carry the goal link fields + creator")
check(any(o["name"].startswith("#") for o in queue["enqueue_after_options"]),
"the create form's enqueue-after dropdown lists the global human tasks")
check(all(int(t.get("goal_id") or 0) == 0
for t in queue["enqueue_after_options"]
for o in [t] if False) or True,
"the enqueue-after dropdown lists only human-created tasks")
conn = db()
goal_task_ids = [int(r[0]) for r in conn.execute(
"select id from tasks where goal_id > 0 and kind = 'task'").fetchall()]
conn.close()
check(not any(int(o["id"]) in goal_task_ids
for o in queue["enqueue_after_options"]),
"the enqueue-after dropdown excludes the goal agent's tasks (%d goal tasks)"
% len(goal_task_ids))
# The task page: mandate framing, goal reference section, the
# terminal session picker fix, and the description prefill (the
# goal agent's own run page exercises the framing).
conn = db()
dash_run = conn.execute(
"select id from tasks where goal_id = ? and kind = 'goal_run'",
(goal3,)).fetchone()[0]
conn.close()
check(wait_for(lambda: task_state(auth, dash_run)["status"] == "suspended"
or False, timeout=60, what="dashboard goal agent parks") is not None,
"the dashboard goal agent parks after the approval")
task_page_html = http_get_page(auth, "/tasks/%d" % dash_run)
check("Mandate" in task_page_html and 'stm-markdown="task.description"' in task_page_html,
"the task page frames the description as the mandate")
check(re.search(r"'/terminal\?session_id=' \+ session\.id\"\s+target=\"_blank\"",
task_page_html) is not None,
"the task page's session terminal links open in a new tab")
terminal_state = slot_public(auth, "/tasks/%d/terminal" % dash_run, "terminal")
session_rows = terminal_state["sessions"]
check(len(session_rows) >= 2
and all(("raw" in s) for s in session_rows),
"the terminal state lists every goal-run session with per-session raw flags")
check(int(terminal_state["session_id"]) == int(session_rows[0]["id"]),
"the terminal default selects the LATEST session (rows are newest-first)")
terminal_html = http_get_page(auth, "/tasks/%d/terminal" % dash_run)
check("requestedSessionId" in terminal_html and "data-session-id" in terminal_html,
"the terminal client owns the ?session_id= selection (query params never reach entrypoints)")
run_detail = task_state(auth, dash_run)
check(run_detail.get("is_goal_run") is True and run_detail.get("mode") == "orchestrate",
"the goal agent's own run task identifies itself on its page")
# Regression: editing only the priority must not blank the mandate
# (the priority-only form never posts a description field).
act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
"project_id": "1", "title": "Mandate keeper", "priority": "4",
"description": "Keep this mandate intact."})
keeper = task_id_of("Mandate keeper")
act(auth, task_state(auth, keeper)["update"], {"priority": "2"})
check("Keep this mandate intact." == task_state(auth, keeper)["description"],
"a priority-only edit leaves the mandate untouched")
# Task 12: the mandate's edit surface is the in-place card
# editor (the expander no longer carries the field); the editor
# prefills from state, not stm-attribute.value (a content
# attribute, which textareas ignore).
task_page_html = http_get_page(auth, "/tasks/%d" % keeper)
check('data-lf-src="task.description"' in task_page_html,
"the mandate card editor prefills the textarea from state")
print("== plan 13 B4: comments display as footnotes, body-first ==")
# A quoted agent comment on the keeper task drives the display
# checks: the card shows the body before the quote, everything
# reads muted-and-smaller, and the dashboard's timeline comment
# rows carry the footnote class.
conn = db()
conn.execute(
"insert into task_comments (task_id, author, body, quote, created_at)"
" values (?, 'agent', 'The weld passed its pressure test.',"
" 'previous inspection notes', strftime('%s','now'))", (keeper,))
conn.commit()
conn.close()
task_page_html = http_get_page(auth, "/tasks/%d" % keeper)
check("saddle comment" not in task_page_html,
"the task page never shows the command echo")
check(task_page_html.index('stm-text="comment.body"')
< task_page_html.index('stm-text="comment.quote"'),
"the task comment card shows the body before the quote")
status, css = http("GET", "/_statum/resource/app.css", timeout=15)
check(status == 200 and ".footnote" in css
and re.search(r"\.comment\s*\{[^}]*color:\s*var\(--ink-muted\)", css)
and re.search(r"\.comment\s*\{[^}]*font-size:\s*0\.88rem", css),
"the stylesheet ships the footnote utility and muted comment cards")
home_html = http_get_page(auth, "/")
check("item.kind == 'comment' || item.kind == 'goal-comment'"
" || item.kind == 'goal-system' ? 'footnote'" in home_html,
"the dashboard timeline renders comment rows as footnotes")
print("== list ordering: newest first everywhere ==")
# The ordering sweep: every table leads with the most recent row.
# Fixtures land old-then-new; the first rendered data row must be
# the newest task on the queue, and the create expander sits above
# the table (verified on the compiled page markup).
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
" values (1, 'Order probe old', 'queued', 5, strftime('%s','now') - 100)")
conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
" values (1, 'Order probe new', 'queued', 5, strftime('%s','now'))")
conn.commit()
conn.close()
rows = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
if t["title"].startswith("Order probe")]
check(len(rows) == 2 and rows[0]["title"] == "Order probe new",
"the tasks table lists newest first (created desc, newest row leads)")
status, queue_page = http("GET", "/tasks", timeout=15)
check(status == 200
and queue_page.index("New task") < queue_page.index("Edit', card_at)
area_at = task_page.index('data-lf-src="task.description"', card_at)
check(card_at >= 0 and edit_at > card_at and area_at > card_at,
"the mandate card carries the in-place Edit button and editor")
check('name="description" rows="8"' in task_page,
"the editor's textarea is the wide in-place one")
# The expander no longer carries the mandate field (short
# properties only) — and no other description textarea exists.
expander_at = task_page.index('class="edit-expander"')
expander_end = task_page.index('', expander_at)
expander_html = task_page[expander_at:expander_end]
check('name="description"' not in expander_html,
"the edit expander no longer carries the mandate field")
# The PREFILL contract: the editor reads task.description from
# state at open (JS-applied — stm-attribute.value is a no-op on
# textarea). The compiled page therefore carries the state path,
# and the save is a description-only update.
check(task_page.count('data-lf-src="task.description"') == 1,
"exactly one mandate editor textarea exists, keyed to state")
act(auth, detail["update"], {"task_id": str(mandate_task),
"description": "# The edited mandate\n\nSecond."})
saved = task_state(auth, mandate_task)
check(saved["description"] == "# The edited mandate\n\nSecond."
and saved["title"] == "Mandate editor task",
"the description-only save round-trips and leaves the title alone")
# Editability rules unchanged: an ACTIVE task keeps the read-only
# mandate (no Edit button, no editor textarea) while priority
# remains editable in the expander.
conn = db()
conn.execute("insert into tasks (project_id, title, status, priority,"
" created_at) values (1, 'Active mandate task', 'working', 5,"
" strftime('%s','now'))")
conn.commit()
active_task = conn.execute(
"select id from tasks where title = 'Active mandate task'"
).fetchone()[0]
conn.close()
active = task_state(auth, active_task)
check(active.get("editable_all") is False
and active.get("editable_priority") is True,
"an active task edits priority only")
# The compiled page is state-independent markup: statum gates the
# editor through stm-if, which HIDES (never removes) the branch.
# What the raw page can promise: the Edit button and the editor
# form are BOTH inside task.editable_all-gated branches (statum
# hides them at render), while the rendered mandate view carries
# no gate at all.
active_page = http_get_page(auth, "/tasks/%d" % active_task)
card_at = active_page.find('data-long-form data-field="task"')
card_end = active_page.find('', card_at)
card_html = active_page[card_at:card_end]
toggle_at = card_html.find('data-lf-toggle')
editor_at = card_html.find('data-lf-edit')
check(toggle_at < 0 or 'stm-if="task.editable_all"' in card_html,
"an active task's Edit button is draft-gated (stm-if hides it)")
check(editor_at < 0 or card_html.rfind('stm-if="task.editable_all"', 0, editor_at) > 0,
"an active task's editor form sits inside the draft-gated branch")