jobbsok/scripts/cowork_probe_check.py

"""Report, from the host, how far the Cowork probe has actually got.

The probe asks four questions that only the operator can answer in a Cowork
session. But the host-MCP one leaves hard traces on this Mac, and a trace is a
measurement where "what did you see in the UI" is a recollection. This script
reads those traces so each step is verified before the next begins.

Python rather than bash: the shell version reached for find, grep, sed, awk and
unzip, and this repository no longer has a shell entry point anywhere. What it
INSPECTS is still macOS-only -- Claude's own log and session directories live
under ~/Library -- so this remains a tool for measuring this Mac. That is a
property of the question, not of the language it is written in.

Two earlier queries in this script were WRONG and the measurement caught both;
the corrections are written down here so they are not re-derived:

  1. It searched for a directory named after the plugin. Cowork installs into
     rpm/plugin_<opaque id>/, so the name never appears in a path. The manifest
     at rpm/manifest.json is the index; read that instead.
  2. It expected ~/Library/Logs/Claude/mcp-server-<name>.log. That naming is
     for Claude Desktop's own connectors. A plugin's MCP server is logged by
     LocalMcpServerManager into main.log instead. No file at the guessed path
     meant the guess was wrong, not that nothing had happened.

It reports; it does not gate. Exit 0 when every check has landed, 1 while any
is pending. Read-only.
"""

import json
import os
import re
import subprocess
import sys
import zipfile

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(SCRIPT_DIR)
PROBE_DOC = os.path.join(REPO_ROOT, "docs", "cowork-probe.md")
FIXTURE = os.path.join(REPO_ROOT, "tests", "fixtures", "cowork-probe")
MAIN_LOG = os.path.expanduser("~/Library/Logs/Claude/main.log")
SESSIONS = os.path.expanduser(
    "~/Library/Application Support/Claude/local-agent-mode-sessions"
)
PLUGIN_NAME = "jobbsok-probe"
SERVER_KEY = "plugin:jobbsok-probe:probe-tools"

#: An archive entry that would mean the probe vehicle leaked something local.
LEKKASJE = re.compile(r"^\.git|STATE\.md|\.venv|^\.claude/")

#: One of the four answers, written down in the form the probe doc asks for.
SVAR = re.compile(
    r"^- (Sesjonsmodus|Host-MCP|python3|CLAUDE_PLUGIN_ROOT): +(lokal VM|sky|ja|nei)\b",
    re.MULTILINE,
)


def lekkasje(navn):
    """True when an archive entry name is one the probe vehicle must not ship."""
    return LEKKASJE.search(navn) is not None


def antall_svar(tekst):
    """How many of the four probe questions are answered in ``tekst``."""
    return len(SVAR.findall(tekst))


def antall_filer(katalog):
    n = 0
    for _dirpath, _dirnames, filenames in os.walk(katalog):
        n += len(filenames)
    return n


def les_tekst(sti):
    """The file's text, or None when it is not there. Never raises on encoding."""
    try:
        with open(sti, "r", encoding="utf-8", errors="replace") as handle:
            return handle.read()
    except IOError:
        return None


def installert_kopi(sessions, navn):
    """Find ``navn`` in any rpm/manifest.json under ``sessions``.

    Returns (directory, id, marketplace, updatedAt) or None. The manifest is
    the index -- the install directory is named after an opaque id, so the
    plugin's own name never appears in a path.
    """
    for dirpath, _dirnames, filenames in os.walk(sessions):
        if os.path.basename(dirpath) != "rpm" or "manifest.json" not in filenames:
            continue
        try:
            with open(os.path.join(dirpath, "manifest.json"), "r", encoding="utf-8") as h:
                data = json.load(h)
        except (IOError, ValueError):
            continue
        for p in data.get("plugins", []):
            if p.get("name") == navn:
                return (
                    os.path.join(dirpath, p["id"]),
                    p["id"],
                    p.get("marketplaceName", "?"),
                    p.get("updatedAt", "?"),
                )
    return None


class Rapport(object):
    def __init__(self):
        self.pending = 0

    def done(self, tekst):
        sys.stdout.write("  [DONE]    %s\n" % tekst)

    def pend(self, tekst):
        sys.stdout.write("  [PENDING] %s\n" % tekst)
        self.pending += 1

    def info(self, tekst):
        sys.stdout.write("            %s\n" % tekst)


def main(argv):
    arkiv = os.environ.get("JOBBSOK_PROBE_ARCHIVE", "/tmp/jobbsok-probe.plugin")
    r = Rapport()

    sys.stdout.write(
        "\n=== 1. Probe vehicle and archive ============================================\n"
    )
    n = antall_filer(FIXTURE)
    if n == 4:
        r.done("probe vehicle: 4 files under tests/fixtures/cowork-probe")
    else:
        r.pend("probe vehicle: expected 4 files, found %d" % n)

    if os.path.isfile(arkiv):
        try:
            with zipfile.ZipFile(arkiv) as pakke:
                oppforinger = pakke.namelist()
        except zipfile.BadZipFile:
            oppforinger = None
        if oppforinger is None:
            r.pend("archive: %s is not a readable zip - rebuild it" % arkiv)
        else:
            lekk = [o for o in oppforinger if lekkasje(o)]
            if len(oppforinger) == 4 and not lekk:
                r.done("archive: %s (4 files, no leaked entries)" % arkiv)
            else:
                r.pend(
                    "archive: %s has %d entries and %d leaked ones - rebuild it"
                    % (arkiv, len(oppforinger), len(lekk))
                )
    else:
        r.pend("archive not built: %s" % arkiv)

    sys.stdout.write(
        "\n=== 2. Is jobbsok-probe installed in Cowork? =================================\n"
    )
    funnet = installert_kopi(SESSIONS, PLUGIN_NAME)
    if funnet:
        katalog, plugin_id, marked, oppdatert = funnet
        r.done(
            "listed in rpm/manifest.json as %s (marketplace: %s, updated %s)"
            % (plugin_id, marked, oppdatert)
        )
        f = antall_filer(katalog)
        if f == 4:
            r.done("installed copy holds 4 files, matching the archive")
        else:
            r.pend("installed copy holds %d files, expected 4" % f)
        manifest = os.path.join(katalog, ".claude-plugin", "plugin.json")
        v = None
        tekst = les_tekst(manifest)
        if tekst:
            try:
                v = json.loads(tekst)["version"]
            except (ValueError, KeyError):
                v = None
        if v == "0.0.1":
            r.done("installed manifest version is 0.0.1 - not a stale cached build")
        else:
            r.pend(
                "installed manifest version is %r, expected 0.0.1 - Cowork served "
                "a cached build" % (v,)
            )
    else:
        r.pend("no plugin named %s in any rpm/manifest.json" % PLUGIN_NAME)

    sys.stdout.write(
        "\n=== 3. Did Cowork spawn probe-tools on THIS Mac? =============================\n"
    )
    sys.stdout.write("    (the -Host-MCP: question, measured rather than recalled)\n")
    logg = les_tekst(MAIN_LOG)
    if logg is not None:
        linjer = logg.splitlines()
        conn = [l for l in linjer if ("Connected to " + SERVER_KEY) in l]
        neg = [l for l in linjer if (SERVER_KEY + " negotiated protocol version") in l]
        if conn:
            r.done("connected: %s" % conn[-1].split("[LocalMcpServerManager] ")[-1])
        else:
            r.pend("main.log has no 'Connected to %s' line" % SERVER_KEY)
        if neg:
            r.done("handshake: negotiated%s" % neg[-1].split("negotiated", 1)[-1])
        else:
            r.pend("no protocol negotiation recorded")
        if [l for l in linjer if "probe_ping" in l]:
            r.done("probe_ping seen in the log")
        else:
            r.info("no probe_ping line in main.log yet - the connection lifecycle is")
            r.info("logged there but an individual tool call may not be. Confirm the")
            r.info("call by its answer in the Cowork chat, not by this line's absence.")
    else:
        r.pend("no main.log at %s" % MAIN_LOG)

    try:
        proc = subprocess.run(
            ["pgrep", "-fl", "probe_tools.py"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
        )
        treff = proc.stdout.splitlines()
    except OSError:
        treff = []
        r.info("no pgrep on this platform - the log above is the record.")
    if treff:
        r.done("process alive: pid %s" % treff[0].split()[0])
        tolk = re.search(r"/[^ ]*python3?", treff[0])
        if tolk:
            r.info("interpreter: %s (host, not sandbox)" % tolk.group(0))
    else:
        r.info("no probe_tools.py process at this instant - servers may be started on")
        r.info("demand, so absence here is not a nei. The log above is the record.")

    sys.stdout.write(
        "\n=== 4. Are the four answers written down? ===================================\n"
    )
    doc = les_tekst(PROBE_DOC) or ""
    c = antall_svar(doc)
    if c == 4:
        r.done("all four answered - plan Step 1 Verify passes")
    else:
        r.pend("%d of 4 answered in docs/cowork-probe.md" % c)
    for linje in doc.splitlines():
        if re.match(r"^- (Sesjonsmodus|Host-MCP|python3|CLAUDE_PLUGIN_ROOT):", linje):
            sys.stdout.write("            %s\n" % linje)

    sys.stdout.write(
        "\n============================================================================\n"
    )
    if r.pending == 0:
        sys.stdout.write("ALL CHECKS LANDED.\n")
        return 0
    sys.stdout.write("%d check(s) still pending.\n" % r.pending)
    return 1


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))