Mark tasks carrying an artifact or a link on the dashboard

README.md

@@ -38,14 +38,14 @@ its `container` status, and its repo and slug:
 ══════════════════════════════════════════════════════════════════════════
   panopticon                                                6 tasks
 ──────────────────────────────────────────────────────────────────────────
-  state          turn       container   repo       slug[memo]
-  ITERATING      agent      live        web-api    add-oauth[Add OAuth login]
-  PLANNING       user       live        web-api    fix-upload[Flaky S3 upload]
-  MERGING        agent      starting    dashboard  dark-mode[Dark-mode theme]
-  ITERATING      user ⚠     down        web-api    migrate-db[Move to Postgres]
-  ORCHESTRATING  agent      live        infra      q3-cleanup[Q3 tech-debt]
-  PLANNING       agent      live        infra      └─ drop-py38[Drop Python 3.8]
-  COMPLETE       agent      –           web-api    ship-readme[README refresh]
+  state          turn       container   repo       ❏ ➚  slug[memo]
+  ITERATING      agent      live        web-api    ❏ ➚  add-oauth[Add OAuth login]
+  PLANNING       user       live        web-api    ❏    fix-upload[Flaky S3 upload]
+  MERGING        agent      starting    dashboard    ➚  dark-mode[Dark-mode theme]
+  ITERATING      user ⚠     down        web-api    ❏ ➚  migrate-db[Move to Postgres]
+  ORCHESTRATING  agent      live        infra      ❏    q3-cleanup[Q3 tech-debt]
+  PLANNING       agent      live        infra           └─ drop-py38[Drop Python 3.8]
+  COMPLETE       agent      –           web-api    ❏ ➚  ship-readme[README refresh]
 ──────────────────────────────────────────────────────────────────────────
   t attach   n new task   x drop   / search   d detail   ? help   q quit
 ══════════════════════════════════════════════════════════════════════════
@@ -55,7 +55,9 @@ The `turn` column is color-coded live: green when the agent is working, yellow w
 move, and red (`⚠`) when a task is blocked waiting on you, so you can tell at a glance which agents
 need you. The `container` column tracks each agent's sandbox as it spawns (`queued → … → live`,
 or `down` when one needs a respawn), and governed sub-tasks nest under their governor (`└─`).
-Press `t` to drop into any task's session, `?` for the full key list.
+The `❏ ➚` column marks what a task carries — `❏` an artifact to read (`a` lists them), `➚` a link
+such as its PR (`p` opens it). Press `t` to drop into any task's session, `?` for the full key
+list.
 
 ## Requirements
 

src/panopticon/core/artifacts.py

@@ -46,6 +46,16 @@ def validate_segment(segment: str) -> None:
         raise InvalidArtifactName(f"invalid artifact segment: {segment!r}")
 
 
+def is_hidden(name: str) -> bool:
+    """Whether an artifact is hidden from the operator's default view.
+
+    Dotfile artifacts are agent bookkeeping — cross-turn state like ``.babysit-ci-state.json`` —
+    rather than documents a human asked for. The dashboard hides them behind a "Show hidden"
+    toggle and the task list's artifact mark ignores them, so the rule lives here rather than
+    being spelled out at each surface."""
+    return name.startswith(".")
+
+
 def mcp_uri(task_id: str, name: str) -> str:
     """The canonical MCP resource URI for an artifact (the shared resolver).
 
@@ -79,6 +89,16 @@ async def get(self, task_id: str, name: str) -> bytes | None:
     async def list(self, task_id: str) -> list[str]:
         """Return the names of a task's artifacts (empty if none)."""
 
+    async def has_unhidden_artifacts(self, task_id: str) -> bool:
+        """Whether the task has at least one artifact the operator would want to open.
+
+        The task list renders a mark per row from this, so it answers the question without
+        materialising names. Concrete, not abstract: the default is written in terms of
+        :meth:`list`, which every adapter must provide, so one that has no cheaper way to tell
+        still inherits a correct implementation (the filesystem store overrides it with a
+        directory scan that stops at the first hit)."""
+        return any(not is_hidden(name) for name in await self.list(task_id))
+
     async def link_slug(self, task_id: str, slug: str) -> None:
         """Expose a task's artifacts under a readable ``slug`` alias (best-effort).
 

src/panopticon/taskservice/api.py

@@ -91,6 +91,11 @@ class TaskSummaryOut(BaseModel):
     sort_weight: int = 0
     depends_on_task_ids: list[str] = []
     provisioned: bool
+    #: Whether the task has at least one unhidden artifact — the dashboard's artifact mark.
+    #: Computed like ``container_status`` (artifacts are files, not a task column), attached on
+    #: serialization by ``_task_summary_out``. Summary-only: the single-task shapes don't carry
+    #: it, since ``GET /tasks/{id}/artifacts`` already answers it exactly for one task.
+    has_artifacts: bool = False
     container_status: str = "–"
     lifecycle_detail: str | None = None
     runner_host: str | None = (
@@ -398,9 +403,13 @@ def _task_out(task: Task) -> TaskOut:
             out.runner_host = service.runner_host(task.claimed_by)
         return out
 
-    def _task_summary_out(task: Task) -> TaskSummaryOut:
-        """Serialize a task to the cheap summary shape (no history), with computed status fields."""
+    def _task_summary_out(task: Task, *, has_artifacts: bool = False) -> TaskSummaryOut:
+        """Serialize a task to the cheap summary shape (no history), with computed status fields.
+
+        ``has_artifacts`` is passed in rather than read here: it lives in the artifact store,
+        not on the task, so resolving it needs an await this synchronous serializer can't do."""
         out = TaskSummaryOut.model_validate(task)
+        out.has_artifacts = has_artifacts
         out.container_status = service.container_status(task).value
         lifecycle = service.lifecycle(task.id)
         out.lifecycle_detail = lifecycle.detail if lifecycle is not None else None
@@ -557,7 +566,10 @@ async def list_tasks(
             # Read version and snapshot in a single thread call so no event-loop yield can
             # interleave a mutation between them — preserving the original atomicity invariant.
             version, tasks_raw = await service._tasks_snapshot(terminal=terminal)
-        tasks = [_task_summary_out(t) for t in tasks_raw]
+        tasks = [
+            _task_summary_out(t, has_artifacts=await service.has_unhidden_artifacts(t.id))
+            for t in tasks_raw
+        ]
         response.headers[TASKS_VERSION_HEADER] = str(version)
         return tasks
 

src/panopticon/taskservice/artifacts_fs.py

@@ -9,9 +9,15 @@
 from __future__ import annotations
 
 import asyncio
+import os
 from pathlib import Path
 
-from panopticon.core.artifacts import ArtifactStore, InvalidArtifactName, validate_segment
+from panopticon.core.artifacts import (
+    ArtifactStore,
+    InvalidArtifactName,
+    is_hidden,
+    validate_segment,
+)
 
 
 class FilesystemArtifactStore(ArtifactStore):
@@ -53,6 +59,30 @@ async def list(self, task_id: str) -> list[str]:
             lambda: sorted(p.name for p in task_dir.iterdir() if p.is_file())
         )
 
+    def _has_artifacts_sync(self, task_id: str) -> bool:
+        """Whether the task's directory holds at least one unhidden file.
+
+        ``os.scandir`` rather than ``Path.iterdir``: it stops at the first hit (the caller wants
+        a boolean, not a listing), its entries answer ``is_file()`` from the data the scan already
+        returned instead of a ``stat`` apiece, and it raises for a missing directory *here* rather
+        than lazily on iteration — ``Path.iterdir`` is a generator before 3.13, so the error would
+        escape this ``try`` on the Python versions we still support.
+        """
+        try:
+            with os.scandir(self._task_dir(task_id)) as entries:
+                return any(not is_hidden(entry.name) and entry.is_file() for entry in entries)
+        except (FileNotFoundError, NotADirectoryError):
+            return False
+
+    async def has_unhidden_artifacts(self, task_id: str) -> bool:
+        """Scan the task's directory directly rather than going through :meth:`list`.
+
+        The inherited default builds and sorts the full name list to answer a question the first
+        unhidden entry settles. The task list asks this for every visible task on every refresh,
+        so the shortcut is worth the override.
+        """
+        return await asyncio.to_thread(self._has_artifacts_sync, task_id)
+
     def _link_slug_sync(self, task_id: str, slug: str) -> None:
         validate_segment(task_id)
         validate_segment(slug)

src/panopticon/taskservice/service.py

@@ -739,6 +739,11 @@ async def record_provisioning(self, task_id: str, *, branch: str, clone: str) ->
     async def put_artifact(self, task_id: str, name: str, content: bytes) -> None:
         await self.get_task(task_id)  # ensure the task exists
         await self._artifacts.put(task_id, name, content)
+        # Artifacts live outside the store, so writing one bumps no version of its own — but the
+        # task list reports whether a task *has* one, so a parked long-poll has to wake or the
+        # first plan.md would go unnoticed until some unrelated mutation. Same treatment as the
+        # other ephemeral (non-stored) changes.
+        self._notify_change()
         _log.debug("task %s: artifact %s written", task_id, name)
 
     async def get_artifact(self, task_id: str, name: str) -> bytes | None:
@@ -749,6 +754,15 @@ async def list_artifacts(self, task_id: str) -> list[str]:
         await self.get_task(task_id)
         return await self._artifacts.list(task_id)
 
+    async def has_unhidden_artifacts(self, task_id: str) -> bool:
+        """Whether the task has an artifact worth marking in the task list.
+
+        No ``get_task`` guard (unlike the readers above): this is a display predicate asked of
+        tasks the caller has already read, once per row, and a task with no artifacts and a task
+        that doesn't exist both answer ``False``. Paying for a store read per row to tell those
+        apart would buy nothing."""
+        return await self._artifacts.has_unhidden_artifacts(task_id)
+
     # -- liveness -----------------------------------------------------------------
     #
     # Liveness is connection-scoped: a container holds the ``/live`` stream open for its whole

src/panopticon/terminal/dashboard.py

@@ -41,6 +41,11 @@
 Arrow keys skip the ensemble row (it is not a real task). Expanding or collapsing does not affect the task service — it is pure
 display state local to the dashboard.
 
+The `❏ ➚` **marks column** (left of `slug[memo]`, its header doubling as the legend) flags what a
+task carries: `❏` when it has at least one unhidden artifact (`a` lists them — dotfile artifacts
+are agent bookkeeping and don't count) and `➚` when it has a `url` (`p` opens it). Each mark keeps
+its own slot, so they read as two vertical rails rather than shifting per row.
+
 The `container` column shows each task's container status: `live` (an active registration), `down`
 (was up, container gone — respawn with `R`), `starting` (claimed, no registration yet — its
 container is still coming up), `healing` (the runner is self-healing an orphan), or `–` (unclaimed
@@ -103,7 +108,7 @@
 from textual.worker import get_current_worker
 
 from panopticon.client import JsonObj, TaskServiceClient
-from panopticon.core.artifacts import InvalidArtifactName, validate_segment
+from panopticon.core.artifacts import InvalidArtifactName, is_hidden, validate_segment
 from panopticon.core.dirs import ARTIFACTS_DIR
 from panopticon.core.models import resolve_agent_cli
 from panopticon.core.state import TERMINAL_LABELS
@@ -187,6 +192,32 @@ def _dim(cell: Text | str) -> Text:
     return t
 
 
+# The task-list marks column: one glyph per slot, separated by a space, so each mark reads down
+# its own vertical line (slot 1 artifact, slot 2 link) and a row missing one doesn't shift the
+# other.
+#
+# Both glyphs are East_Asian_Width=Neutral with no emoji presentation form, i.e. exactly one cell
+# in every terminal. That is the whole reason for these two rather than 📁/🔗: emoji are
+# Width=Wide (two cells, and fonts disagree on the details), which would make the column's width
+# depend on the viewer. Keep any replacement in the same class — `⚠`/`✓` elsewhere in this file
+# are the other examples.
+_ARTIFACT_MARK = "❏"  # U+274F, has at least one unhidden artifact (`a` lists them)
+_LINK_MARK = "➚"  # U+279A, has a url (`p` opens it)
+_MARKS_LABEL = f"{_ARTIFACT_MARK} {_LINK_MARK}"  # the header doubles as the legend
+_MARKS_HEADER = Text(_MARKS_LABEL)
+
+
+def _marks_cell(task: JsonObj) -> Text:
+    """The marks column: ``❏`` when the task has an unhidden artifact, ``➚`` when it has a url.
+
+    Always three cells wide (mark, gap, mark) — an absent mark renders as a space rather than
+    collapsing — so the two marks stay in their own columns and the cell width can't vary by
+    row."""
+    artifact = _ARTIFACT_MARK if task.get("has_artifacts") else " "
+    link = _LINK_MARK if task.get("url") else " "
+    return Text(f"{artifact} {link}", style="dim")
+
+
 def _slug_cell(task: JsonObj, prefix: str = "") -> Text:
     """The ``slug[memo]`` column: the slug followed by the task's memo in brackets.
 
@@ -1721,22 +1752,18 @@ class ArtifactScreen(_OptionListModal[tuple[str, str]]):
 
     def __init__(self, title: str, all_names: list[str]) -> None:
         self._all_names = all_names
-        visible = [n for n in all_names if not n.startswith(".")]
+        visible = [n for n in all_names if not is_hidden(n)]
         super().__init__(title, visible)
 
     def _extra_widgets(self) -> Iterable[Widget]:
         yield Label(
             "enter: open · e: open local file · ctrl+a: attach · esc: cancel", id="artifact-hint"
         )
-        if any(n.startswith(".") for n in self._all_names):
+        if any(is_hidden(n) for n in self._all_names):
             yield SpaceCheckbox("Show hidden", id="show-hidden")
 
     def on_checkbox_changed(self, event: Checkbox.Changed) -> None:
-        names = (
-            self._all_names
-            if event.value
-            else [n for n in self._all_names if not n.startswith(".")]
-        )
+        names = self._all_names if event.value else [n for n in self._all_names if not is_hidden(n)]
         option_list = self.query_one(OptionList)
         option_list.clear_options()
         for name in names:
@@ -1906,11 +1933,16 @@ def action_close(self) -> None:
 
 
 def _setup_task_columns(table: DataTable[Any], *, multi_runner: bool) -> None:
-    """Add the task table's columns. Includes a "runner" column when tasks span multiple hosts."""
+    """Add the task table's columns. Includes a "runner" column when tasks span multiple hosts.
+
+    The marks column sits immediately left of the name it annotates, and last among the
+    fixed-width columns so the variable-width ``slug[memo]`` stays rightmost."""
     if multi_runner:
-        table.add_columns("state", "turn", "container", "runner", "repo", Text("slug[memo]"))
+        table.add_columns(
+            "state", "turn", "container", "runner", "repo", _MARKS_HEADER, Text("slug[memo]")
+        )
     else:
-        table.add_columns("state", "turn", "container", "repo", Text("slug[memo]"))
+        table.add_columns("state", "turn", "container", "repo", _MARKS_HEADER, Text("slug[memo]"))
 
 
 class Dashboard(App[None]):
@@ -2140,6 +2172,7 @@ def _add_row(task: JsonObj, prefix: str) -> None:
                     Text(""),
                     *runner_blank,
                     Text(""),
+                    Text(""),  # marks: a placeholder stands for hidden rows, so it carries none
                     slug_cell,
                     key=f"{_ENSEMBLE_KEY_PREFIX}{gov_id}",
                 )
@@ -2151,6 +2184,7 @@ def _add_row(task: JsonObj, prefix: str) -> None:
                     Text(task.get("runner_host") or "") if self._multi_runner else None
                 )
                 repo_cell: Text | str = _repo_cell(task, self._repo_names)
+                marks_cell = _marks_cell(task)
                 slug_cell_real = _slug_cell(task, prefix)
                 if task["state"] in TERMINAL_LABELS:
                     state_cell = _dim(state_cell)
@@ -2159,6 +2193,7 @@ def _add_row(task: JsonObj, prefix: str) -> None:
                     if runner_cell is not None:
                         runner_cell = _dim(runner_cell)
                     repo_cell = _dim(repo_cell)
+                    marks_cell = _dim(marks_cell)
                     slug_cell_real = _dim(slug_cell_real)
                 elif _snooze_label(task, display_now) is not None:
                     # An active snooze mutes the whole row (the turn cell already carries the label).
@@ -2168,6 +2203,7 @@ def _add_row(task: JsonObj, prefix: str) -> None:
                     if runner_cell is not None:
                         runner_cell = _dim(runner_cell)
                     repo_cell = _dim(repo_cell)
+                    marks_cell = _dim(marks_cell)
                     slug_cell_real = _dim(slug_cell_real)
                 runner_extra = (runner_cell,) if runner_cell is not None else ()
                 table.add_row(
@@ -2176,6 +2212,7 @@ def _add_row(task: JsonObj, prefix: str) -> None:
                     status_cell,
                     *runner_extra,
                     repo_cell,
+                    marks_cell,
                     slug_cell_real,
                     key=task["id"],
                 )

tests/taskservice/test_api.py

@@ -550,3 +550,26 @@ def test_report_unknown_responsibility(gated_client: TestClient) -> None:
         f"/tasks/{task_id}/responsibilities", json={"key": "ghost", "status": "met"}
     )
     assert resp.status_code == 400
+
+
+def test_task_list_reports_whether_a_task_has_unhidden_artifacts(client: TestClient) -> None:
+    # The dashboard renders a mark per row, so presence has to ride along on the list response
+    # rather than costing a request per task. Dotfile artifacts are agent bookkeeping: they
+    # don't count as something the operator can open.
+    def make() -> str:
+        task_id: str = client.post("/tasks", json={"repo_id": "r1", "workflow": "spike"}).json()[
+            "id"
+        ]
+        return task_id
+
+    visible, hidden_only, bare = make(), make(), make()
+    assert client.put(f"/tasks/{visible}/artifacts/plan.md", content=b"# Plan").status_code == 204
+    assert (
+        client.put(
+            f"/tasks/{hidden_only}/artifacts/.babysit-ci-state.json", content=b"{}"
+        ).status_code
+        == 204
+    )
+
+    marks = {t["id"]: t["has_artifacts"] for t in client.get("/tasks").json()}
+    assert marks == {visible: True, hidden_only: False, bare: False}