okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petfeat: background cells, jobs page, and session list fixesf65a6482026-10-03

"""The vent channel must record model complaints for user review, share one
global ledger across sessions and workspaces, and let /paperclips triage
reach the model — with a reply recorded on the vent and delivered to the
session that filed it.

Each scenario has an isolated daemon because the model's bounded global
listing has no page cursor. Sessions within a scenario still share one ledger."""

import ast
import json
import sqlite3
import unittest

from harness import Albedo, Provider, exclusive, python, text

VENT = (
    "v = await vent('harness', 'a background job stopped responding',"
    " suggestion='report the blocked job', title='job stalled')\n"
    "print(v['id'])"
)

READ_BACK = "print([(v['id'], v['status']) for v in await vents()])"

# Prints the vent this session filed, then the whole global listing: the
# second line proves another session's vent is visible to the model.
VENT_AND_LIST = (
    "v = await vent('harness', 'a background job stopped responding',"
    " suggestion='report the blocked job', title='job stalled')\n"
    "print(v['id'])\n"
    "print([v['id'] for v in await vents()])"
)

READER_TURN = "print('the reader turns')"

# Refused without a note, closed with one, then refused as already closed.
RESOLVE = (
    "vent_id = v['id']\n"
    "for note in ('  ', 'the blocked job now reports its state', 'again'):\n"
    "    try:\n"
    "        v = await resolve_vent(vent_id, note)\n"
    "        print('resolved:', v['status'], v['resolution'])\n"
    "    except Exception as error:\n"
    "        print('refused:', error)"
)

TITLELESS = (
    "v = await vent('bug', 'wires crossed somewhere deep in the stack')\nprint(v['id'])"
)

REJECTED = (
    "try:\n"
    "    await vent('nonsense', 'no such topic')\n"
    "except Exception as error:\n"
    "    print('rejected:', error)\n"
    "try:\n"
    "    await vent('bug', '   ')\n"
    "except Exception as error:\n"
    "    print('rejected:', error)\n"
    "print([v['id'] for v in await vents()"
    " if v['message'] in ('no such topic', '   ')])"
)


def scripting(cells):
    """Answers each user turn with the next scripted python cell."""

    def reply(request):
        if request["messages"][-1].get("role") == "user" and cells:
            return python(cells.pop(0))
        return text("done")

    return Provider(reply)


def python_results(app, session):
    return [
        json.loads(part["value"])
        for entry in app.history(session)["items"]
        if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python"
        for part in entry["content"]
        if part["kind"] == "json" and part["field"] == "result"
    ]


def filed_id(app, session):
    """The vent id the session's first cell printed."""
    output = python_results(app, session)[0]["output"]
    return int(output.strip())


def statuses(app, session):
    """The id -> status map the session's latest listing printed."""
    output = python_results(app, session)[-1]["output"]
    return dict(ast.literal_eval(output.strip()))


def vent_row(page, vent_id):
    return next(
        resource["value"]
        for resource in page["items"]
        if resource["value"]["id"] == str(vent_id)
    )


def sidebar_extensions(app, session):
    with app.api(f"/sessions/{session}?tail=0") as response:
        glances = json.load(response)["glances"]
    return {glance["extension"] for glance in glances}


def paperclips_page(app):
    with app.api("/extensions/paperclips/items?limit=200") as response:
        return json.load(response)


def reply_to_vent(app, vent_id, message):
    route = f"/extensions/paperclips/items/{vent_id}"
    with app.api(route) as response:
        json.load(response)
        revision = response.headers["ETag"]
    with app.api(
        route, {"reply": message}, method="PATCH", headers={"If-Match": revision}
    ) as response:
        return json.load(response)


def rename(app, session, name):
    with app.api(f"/sessions/{session}?view=configuration") as response:
        json.load(response)
        revision = response.headers["ETag"]
    app.api(
        f"/sessions/{session}?view=configuration",
        {"name": name},
        method="PATCH",
        headers={"If-Match": revision},
    ).close()


@exclusive
class PaperclipsTests(unittest.TestCase):
    def remove_vents(self, app, session, ids):
        """Keeps one test's vents out of the next test's global listing."""
        for vent_id in ids:
            route = f"/extensions/paperclips/items/{vent_id}"
            with app.api(route) as response:
                json.load(response)
                revision = response.headers["ETag"]
            app.api(route, method="DELETE", headers={"If-Match": revision}).close()

    def test_vents_share_one_global_ledger(self):
        provider = scripting([VENT, VENT_AND_LIST])
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            sessions = []
            ids = []
            try:
                outputs = {}
                for name in ("vent-first", "vent-second"):
                    workspace = app.root / name
                    workspace.mkdir()
                    session = app.session(workspace)
                    sessions.append(session)
                    app.prompt(session, "vent about the build").close()
                    app.idle(session)
                    results = python_results(app, session)
                    self.assertEqual(len(results), 1)
                    self.assertEqual(results[0]["status"], "ok", results[0])
                    outputs[name] = results[0]["output"].strip().splitlines()
                    ids.append(int(outputs[name][0]))
                # the second session's python list already carries the
                # first's vent: the model sees every session's vents
                listed = ast.literal_eval(outputs["vent-second"][1])
                self.assertIn(ids[0], listed)
                # and the first session's own /paperclips page shows both
                page = paperclips_page(app)
                rows = {item["value"]["id"] for item in page["items"]}
                self.assertTrue({str(vent_id) for vent_id in ids} <= rows)
                # open vents stay out of the chat sidebar
                self.assertNotIn("paperclips", sidebar_extensions(app, sessions[0]))
            finally:
                if ids:
                    self.remove_vents(app, sessions[0], ids)

    def test_user_triage_reaches_the_model(self):
        provider = scripting([VENT, READ_BACK])
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            workspace = app.root / "vent-triage"
            workspace.mkdir()
            session = app.session(workspace)
            vent_id = None
            try:
                app.prompt(session, "vent about the build").close()
                app.idle(session)
                vent_id = filed_id(app, session)

                # a named session shows by name in the detail pane
                rename(app, session, "grumpy-venter")

                page = paperclips_page(app)
                # the list shows the model's title; the detail carries the vent
                row = vent_row(page, vent_id)
                self.assertEqual(row["title"], "job stalled")
                self.assertEqual(row["status"], "open")
                self.assertEqual(row["message"], "a background job stopped responding")
                self.assertEqual(row["suggestion"], "report the blocked job")
                self.assertEqual(row["session_id"], session)

                replied = reply_to_vent(app, vent_id, "the blocked job is now visible")
                self.assertEqual(replied["notification"]["state"], "queued")
                self.assertEqual(replied["resource"]["value"]["status"], "acknowledged")
                self.assertEqual(
                    replied["resource"]["value"]["reply"],
                    "the blocked job is now visible",
                )

                page = paperclips_page(app)
                self.assertEqual(vent_row(page, vent_id)["status"], "acknowledged")
                self.assertIn(
                    "the blocked job is now visible",
                    vent_row(page, vent_id)["reply"],
                )

                # the note reaches the model with its next turn; the ledger agrees
                app.prompt(session, "check your vents").close()
                app.idle(session)
                results = python_results(app, session)
                self.assertEqual(len(results), 2)
                self.assertEqual(statuses(app, session)[vent_id], "acknowledged")
                # the queued note reaches the model as its own user turn
                self.assertTrue(
                    any(
                        "answers: the blocked job is now visible"
                        in str(message.get("content"))
                        for request in provider.requests
                        for message in request["request"]["messages"]
                    )
                )
            finally:
                if vent_id is not None:
                    self.remove_vents(app, session, [vent_id])

    def test_a_reply_reaches_the_session_that_vented(self):
        provider = scripting([VENT, READ_BACK, READER_TURN])
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            sessions = {}
            for name in ("vent-filer", "vent-reader"):
                workspace = app.root / name
                workspace.mkdir()
                sessions[name] = app.session(workspace)
            filer = sessions["vent-filer"]
            reader = sessions["vent-reader"]
            vent_id = None
            try:
                app.prompt(filer, "vent about the build").close()
                app.idle(filer)
                vent_id = filed_id(app, filer)

                # the reader sees the filer's vent, attributed to its workspace
                page = paperclips_page(app)
                row = vent_row(page, vent_id)
                self.assertEqual(row["status"], "open")
                self.assertEqual(row["session_id"], filer)
                self.assertIn("vent-filer", row["workspace"])

                replied = reply_to_vent(app, vent_id, "the reader answers")
                self.assertEqual(replied["notification"]["state"], "queued")
                self.assertEqual(replied["resource"]["value"]["status"], "acknowledged")

                # the note lands on the filer's next turn...
                app.prompt(filer, "check your vents").close()
                app.idle(filer)
                results = python_results(app, filer)
                self.assertEqual(len(results), 2)
                self.assertEqual(statuses(app, filer)[vent_id], "acknowledged")
                self.assertIn(
                    "answers: the reader answers",
                    str(provider.requests[-1]["request"]["messages"]),
                )
                # ...and not on the reader's, which turns afterwards. Each
                # python turn spans two requests (tool call, then its
                # result), so the requests carrying the note are identified
                # by the filer's prompt, not by position.
                app.prompt(reader, "anything for me?").close()
                app.idle(reader)
                noted = [
                    request
                    for request in provider.requests
                    if "answers: the reader answers"
                    in str(request["request"]["messages"])
                ]
                self.assertTrue(noted)
                for request in noted:
                    self.assertIn(
                        "check your vents",
                        str(request["request"]["messages"]),
                    )
                    self.assertNotIn(
                        "anything for me?",
                        str(request["request"]["messages"]),
                    )
            finally:
                if vent_id is not None:
                    self.remove_vents(app, reader, [vent_id])

    # exclusive: restarts the daemon to load seeded vents
    @exclusive
    def test_an_undeliverable_reply_is_kept_on_the_vent(self):
        """A reply survives its note: the answer is durable on the vent even
        when the session that filed it is gone or was never recorded."""
        provider = Provider(lambda request: text("done"))
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            session = app.session()
            database = app.home / "albedo.sqlite"

            def prepare(_app):
                with sqlite3.connect(database) as db:
                    db.execute(
                        "INSERT INTO paperclips(topic,message,cwd,session) "
                        "VALUES('bug','ghost vent',?,'no-such-session')",
                        (str(app.workspace),),
                    )
                    db.execute(
                        "INSERT INTO paperclips(topic,message,cwd) "
                        "VALUES('bug','sessionless vent',?)",
                        (str(app.workspace),),
                    )

            app.restart(prepare=prepare)
            ids = {}
            try:
                page = paperclips_page(app)
                for text_label in ("ghost vent", "sessionless vent"):
                    row = next(
                        item["value"]
                        for item in page["items"]
                        if item["value"]["message"] == text_label
                    )
                    self.assertEqual(row["status"], "open")
                    ids[text_label] = int(row["id"])

                gone = reply_to_vent(app, ids["ghost vent"], "for the ghost")
                self.assertNotEqual(gone["notification"]["state"], "queued")
                self.assertEqual(gone["resource"]["value"]["status"], "acknowledged")
                self.assertEqual(gone["resource"]["value"]["reply"], "for the ghost")

                nobody = reply_to_vent(app, ids["sessionless vent"], "for nobody")
                self.assertNotEqual(nobody["notification"]["state"], "queued")
                self.assertEqual(nobody["resource"]["value"]["status"], "acknowledged")
                self.assertEqual(nobody["resource"]["value"]["reply"], "for nobody")

                # the answers are on the vents, and neither stays in the glance
                page = paperclips_page(app)
                self.assertIn(
                    "for the ghost",
                    vent_row(page, ids["ghost vent"])["reply"],
                )
                self.assertIn(
                    "for nobody",
                    vent_row(page, ids["sessionless vent"])["reply"],
                )
            finally:
                self.remove_vents(app, session, list(ids.values()))

    def test_the_model_resolves_a_vent_it_fixed_with_a_note(self):
        provider = scripting([VENT, RESOLVE])
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            session = app.session()
            vent_id = None
            try:
                app.prompt(session, "vent about the build").close()
                app.idle(session)
                vent_id = filed_id(app, session)
                app.prompt(session, "you fixed it, close it").close()
                app.idle(session)

                output = python_results(app, session)[-1]["output"]
                lines = output.strip().splitlines()
                self.assertEqual(len(lines), 3, output)
                self.assertIn("refused: say what fixed the vent", lines[0])
                self.assertEqual(
                    lines[1], "resolved: resolved the blocked job now reports its state"
                )
                self.assertIn(f"refused: vent #{vent_id} is already resolved", lines[2])

                # the user sees the closure and who made it; it leaves the glance
                rename(app, session, "fixer")
                page = paperclips_page(app)
                row = vent_row(page, vent_id)
                self.assertEqual(row["status"], "resolved")
                self.assertEqual(row["resolving_session_id"], session)
                self.assertEqual(
                    row["resolution"], "the blocked job now reports its state"
                )
            finally:
                if vent_id is not None:
                    self.remove_vents(app, session, [vent_id])

    def test_a_titleless_vent_takes_its_title_from_the_message(self):
        provider = scripting([TITLELESS])
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            session = app.session()
            vent_id = None
            try:
                app.prompt(session, "vent").close()
                app.idle(session)
                vent_id = filed_id(app, session)
                row = vent_row(paperclips_page(app), vent_id)
                self.assertEqual(
                    row["title"], "wires crossed somewhere deep in the stack"
                )
                self.assertIn("wires crossed", row["message"])
            finally:
                if vent_id is not None:
                    self.remove_vents(app, session, [vent_id])

    def test_a_vent_needs_a_known_topic_and_a_message(self):
        provider = scripting([REJECTED])
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            session = app.session()
            app.prompt(session, "vent something").close()
            app.idle(session)
            results = python_results(app, session)
            self.assertEqual(results[0]["status"], "ok", results[0])
            # both attempts were rejected — the whitespace message included
            self.assertEqual(results[0]["output"].count("rejected:"), 2)
            # and neither left a vent in the global ledger
            self.assertTrue(results[0]["output"].strip().endswith("[]"))