Workers may file work items; new items always land in the proposed gate

Worker slice now includes items it created, so a filed follow-up stays visible to its author.

coworker/teams/store.py

@@ -108,6 +108,7 @@ class TeamStore:
                 criteria TEXT NOT NULL,
                 state TEXT NOT NULL,
                 assignee TEXT DEFAULT '',
+                creator TEXT NOT NULL DEFAULT '',
                 case_id TEXT DEFAULT '',
                 refs TEXT NOT NULL DEFAULT '[]',
                 created_ts TEXT NOT NULL,
@@ -215,7 +216,7 @@ class TeamStore:
             ),
         )
         seq = cursor.lastrowid
-        self._apply(space, seq, record["ts"], kind, item_id, payload)
+        self._apply(space, seq, record["ts"], kind, actor.id, item_id, payload)
         self._conn.execute(
             """
             INSERT INTO team_meta (space, head_hash, watermark) VALUES (?, ?, ?)
@@ -306,7 +307,7 @@ class TeamStore:
             self._conn.execute("DELETE FROM team_items WHERE space = ?", (space,))
             self._conn.execute("DELETE FROM team_links WHERE space = ?", (space,))
             rows = self._conn.execute(
-                "SELECT seq, ts, kind, item_id, payload FROM team_events"
+                "SELECT seq, ts, kind, actor, item_id, payload FROM team_events"
                 " WHERE space = ? ORDER BY seq",
                 (space,),
             ).fetchall()
@@ -316,6 +317,7 @@ class TeamStore:
                     row["seq"],
                     row["ts"],
                     row["kind"],
+                    row["actor"],
                     row["item_id"],
                     json.loads(row["payload"]),
                 )
@@ -334,8 +336,12 @@ class TeamStore:
         parent: Optional[int] = None,
         case: Optional[str] = None,
     ) -> dict[str, Any]:
-        """New item in `proposed`. Acceptance criteria are load-bearing — required."""
-        self._require(actor, {Role.USER, Role.LEAD}, "create_item")
+        """New item in `proposed`. Acceptance criteria are load-bearing — required.
+
+        Workers may create too — a bug spotted in passing, a follow-up — because
+        proposing is harmless: nothing runs until the item crosses the approval
+        gate."""
+        self._require(actor, {Role.USER, Role.LEAD, Role.WORKER}, "create_item")
         if not (title or "").strip():
             raise BoardError("title is required")
         if not (criteria or "").strip():
@@ -666,6 +672,7 @@ class TeamStore:
         seq: int,
         ts: str,
         kind: str,
+        actor_id: str,
         item_id: Optional[int],
         payload: dict[str, Any],
     ) -> None:
@@ -677,8 +684,8 @@ class TeamStore:
                 """
                 INSERT INTO team_items
                     (space, id, title, description, criteria, state, assignee,
-                     case_id, refs, created_ts, updated_seq)
-                VALUES (?, ?, ?, ?, ?, ?, '', ?, '[]', ?, ?)
+                     creator, case_id, refs, created_ts, updated_seq)
+                VALUES (?, ?, ?, ?, ?, ?, '', ?, ?, '[]', ?, ?)
                 """,
                 (
                     space,
@@ -687,6 +694,7 @@ class TeamStore:
                     payload.get("description") or "",
                     payload.get("criteria") or "",
                     ItemState.PROPOSED.value,
+                    actor_id,
                     payload.get("case") or "",
                     ts,
                     seq,
@@ -782,9 +790,12 @@ class TeamStore:
             )
 
     def _worker_slice(self, space: str, worker_id: str) -> set[int]:
+        # Assigned items, items the worker filed itself, and items directly
+        # linked to either — its slice of the board, nothing more.
         rows = self._conn.execute(
-            "SELECT id FROM team_items WHERE space = ? AND assignee = ?",
-            (space, worker_id),
+            "SELECT id FROM team_items WHERE space = ?"
+            " AND (assignee = ? OR creator = ?)",
+            (space, worker_id, worker_id),
         ).fetchall()
         mine = {row["id"] for row in rows}
         if not mine:

coworker/teams/tools.py

@@ -19,7 +19,9 @@ from .model import Actor, BoardError, Role
 from .store import TeamStore
 
 LEAD_VERBS = ("create_item", "list_items", "transition", "comment", "assign", "link")
-WORKER_VERBS = ("list_items", "transition", "comment")
+# Workers file items too (a bug spotted in passing, a follow-up) — new items land
+# in `proposed`, so the approval gate catches everything a worker proposes.
+WORKER_VERBS = ("create_item", "list_items", "transition", "comment")
 JOURNAL_VERBS = ("journal_append", "journal_read")
 
 # Explicit schema: the auto-generator's normalizer strips every `title` key to drop

tests/test_team_board.py

@@ -33,9 +33,21 @@ def test_acceptance_criteria_are_required(store):
         store.create_item(SPACE, LEAD, title="Vague hope", criteria="  ")
 
 
-def test_workers_cannot_create_items(store):
-    with pytest.raises(AuthorityError):
-        store.create_item(SPACE, WORKER, title="Scope creep", criteria="c")
+def test_workers_file_items_into_the_proposed_gate(store):
+    mine = assigned_item(store)
+    filed = store.create_item(
+        SPACE, WORKER, title="Rounding bug in invoices", criteria="repro + fix",
+        parent=mine,
+    )
+    assert filed["state"] == "proposed"
+    assert filed["creator"] == "worker-1"
+    # the worker sees its own filing; nothing runs until the user approves it
+    visible = {item["id"] for item in store.list_items(SPACE, WORKER)}
+    assert filed["id"] in visible
+    with pytest.raises(AuthorityError, match="decomposition gate"):
+        store.transition(SPACE, WORKER, filed["id"], "approved")
+    other_worker = {item["id"] for item in store.list_items(SPACE, OTHER)}
+    assert filed["id"] not in other_worker
 
 
 def test_child_inherits_parent_case(store):
@@ -197,7 +209,7 @@ def test_tool_sets_are_role_filtered(store):
         tool.__name__ for tool in board_tools(store, space=SPACE, actor=WORKER)
     }
     assert lead_names == {"create_item", "list_items", "transition", "comment", "assign", "link"}
-    assert worker_names == {"list_items", "transition", "comment"}
+    assert worker_names == {"create_item", "list_items", "transition", "comment"}
 
 
 def test_tools_return_errors_instead_of_raising(store):