Projects: beyond-parity queue WS-27u–z + Tasks↔Projects UI continuity

apps/services/gateway/gateway/routes/projects/__init__.py

@@ -29,6 +29,7 @@
 from gateway.routes.projects import custom_fields as _custom_fields  # noqa: F401
 from gateway.routes.projects import import_clickup as _import_clickup  # noqa: F401
 from gateway.routes.projects import import_tasks as _import_tasks  # noqa: F401
+from gateway.routes.projects import intake as _intake  # noqa: F401
 from gateway.routes.projects import me as _me  # noqa: F401
 from gateway.routes.projects import notifications as _notifications  # noqa: F401
 from gateway.routes.projects import personal as _personal  # noqa: F401
@@ -39,6 +40,7 @@
 from gateway.routes.projects import tasks as _tasks  # noqa: F401
 from gateway.routes.projects import tree as _tree  # noqa: F401
 from gateway.routes.projects import views as _views  # noqa: F401
+from gateway.routes.projects import watchers as _watchers  # noqa: F401
 from gateway.routes.projects.core import router
 
 __all__ = ["router"]

apps/services/gateway/gateway/routes/projects/activities.py

@@ -31,6 +31,7 @@
     load_visible_task,
     now,
     record_activity,
+    record_field_change,
     resolve_visibility,
     router,
     row_to_dict,
@@ -39,9 +40,11 @@
 from gateway.routes.projects.notifications import (
     excerpt_of,
     mention_targets,
+    new_mentions,
     notify,
     task_audience,
 )
+from gateway.routes.projects.watchers import ensure_watchers
 from pydantic import BaseModel
 from sqlalchemy import text
 
@@ -128,17 +131,27 @@ async def add_comment(
             db, activity_type="comment", created_by=actor(user),
             task_id=task_id, body=body,
         )
+        # WS-27v — commenting subscribes the commenter, BEFORE the audience is
+        # read so their row exists for the next event too. Harmless for this
+        # one: `notify` never addresses the actor (rule 1).
+        await ensure_watchers(db, task_id, [actor(user)], by=actor(user))
         # WS-27j. Two audiences, and the order matters: a person who is BOTH
-        # mentioned and an assignee should get the mention, because "you were
-        # named" is a stronger claim on their attention than "the task you hold
-        # got a comment". `notifiable` dedupes within a call, so the assignee
-        # pass is handed only whoever the mention pass did not take.
+        # mentioned and a watcher/assignee should get the mention, because "you
+        # were named" is a stronger claim on their attention than "the task you
+        # follow got a comment". `notifiable` dedupes within a call, so the
+        # audience pass is handed only whoever the mention pass did not take.
         mentioned = mention_targets(body)
         snippet = excerpt_of(body)
         mention_result = await notify(
             db, recipients=mentioned, kind="mention", task_id=task_id,
             actor_id=actor(user), activity_id=str(row.id), excerpt=snippet,
         )
+        # Being @mentioned subscribes you — but only the DELIVERED mentions:
+        # a person who cannot open the task must not be silently enrolled in a
+        # stream of its titles they could never have asked for.
+        await ensure_watchers(
+            db, task_id, mention_result["notified"], by=actor(user),
+        )
         rest = [
             who for who in await task_audience(db, task_id)
             if who.strip().lower() not in set(mentioned)
@@ -206,8 +219,35 @@ async def edit_comment(
         vis = await resolve_visibility(db, user)
         await load_visible_task(db, vis, str(comment.task_id))
         row = await update_row(db, "pm_activities", activity_id, {"body": body})
+        task_id = str(comment.task_id)
+        # WS-27v — mention DIFFING: only the addresses this edit ADDED are
+        # notified. The old body is the one loaded above, before the update, so
+        # fixing a typo next to `@priya@…` re-pings nobody, while appending
+        # `@ravi@…` reaches exactly Ravi. Editing your own comment is a touch,
+        # so the editor (re-)subscribes too — idempotently, since they already
+        # subscribed when they commented.
+        added = new_mentions(getattr(comment, "body", None), body)
+        mention_result = await notify(
+            db, recipients=added, kind="mention", task_id=task_id,
+            actor_id=actor(user), activity_id=activity_id,
+            excerpt=excerpt_of(body),
+        )
+        await ensure_watchers(
+            db, task_id, [actor(user), *mention_result["notified"]],
+            by=actor(user),
+        )
+        if mention_result["notified"]:
+            # On the timeline, as add_comment does: why a colleague turned up
+            # must be readable by everyone, not only in one person's bell.
+            await record_activity(
+                db, activity_type="mention", created_by=actor(user),
+                task_id=task_id,
+                meta={"mentioned": mention_result["notified"]},
+            )
         await db.commit()
-        return row_to_dict(row, ActivityModel)
+        result = row_to_dict(row, ActivityModel)
+        result["not_notified"] = mention_result["skipped"]
+        return result
     finally:
         await db.close()
 
@@ -323,20 +363,20 @@ async def revert_change(
             )
 
         await update_row(db, "pm_tasks", str(task.id), restore)
-        await record_activity(
-            db, activity_type="field_change", created_by=actor(user),
-            task_id=str(task.id),
-            meta={
-                "changes": [
-                    {"field": f, "old": c.get("new"), "new": c.get("old")}
-                    for c in changes for f in [c.get("field")]
-                    if f in restore or (
-                        custom_restored is not None
-                        and str(f or "").startswith(_CUSTOM_PREFIX)
-                    )
-                ],
-                "reverted_activity_id": activity_id,
-            },
+        # The ONE field_change door (WS-27w). `extra_meta` marks this as a
+        # revert, which also exempts it from description coalescing — undoing
+        # an edit must never be folded into the edit it undoes.
+        await record_field_change(
+            db, created_by=actor(user), task_id=str(task.id),
+            changes=[
+                {"field": f, "old": c.get("new"), "new": c.get("old")}
+                for c in changes for f in [c.get("field")]
+                if f in restore or (
+                    custom_restored is not None
+                    and str(f or "").startswith(_CUSTOM_PREFIX)
+                )
+            ],
+            extra_meta={"reverted_activity_id": activity_id},
         )
         await db.commit()
         task_id = str(task.id)

apps/services/gateway/gateway/routes/projects/automation.py

@@ -35,16 +35,25 @@
 
 from __future__ import annotations
 
+from calendar import monthrange
+from datetime import datetime, time
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from gateway.routes.projects.core import (
+    CLOSING_CATEGORIES,
+    TRIAGE_CATEGORY,
     apply_status_transition,
     coerce_write_values,
     diff_changes,
     record_activity,
+    record_field_change,
     require_row,
     update_row,
 )
+from gateway.routes.projects.core import (
+    now as _clock,  # aliased: the sweep's own parameter is called `now`
+)
 from sqlalchemy import text
 
 #: The fields an automation may patch. A deliberately SHORT list: everything
@@ -154,17 +163,22 @@ async def apply_task_patch(
             diffs = diff_changes(task, after, _TRACKED)
             changed = [str(d["field"]) for d in diffs]
             if diffs:
-                await record_activity(
-                    db, activity_type="field_change", created_by=actor,
-                    task_id=task_id, meta={"changes": diffs},
+                # The ONE field_change door (WS-27w): an automation's edit gets
+                # the same label resolution and description coalescing a human
+                # PATCH gets — a workflow that rewrites a description every run
+                # must not write a timeline row every run. Flagged automation
+                # (WS-27z) so the timeline renders it distinctly.
+                await record_field_change(
+                    db, created_by=actor, task_id=task_id, changes=diffs,
+                    automation=True,
                 )
 
     status_name: str | None = None
     if wanted_status is not None:
         target = await resolve_status(db, str(after.root_project_id), str(wanted_status))
         if str(target.id) != str(after.status_id):
             moved = await apply_status_transition(
-                db, after, str(target.id), created_by=actor,
+                db, after, str(target.id), created_by=actor, automation=True,
             )
             after = moved["row"]
             changed.append("status")
@@ -191,3 +205,188 @@ def _differs(current: Any, wanted: Any) -> bool:
     if current is None or wanted is None:
         return current is not wanted
     return str(current) != str(wanted)
+
+
+# ── The lifecycle sweep (WS-27z) ────────────────────────────────────────────
+#
+# Policy lives in migration 166's three ROOT-project columns; the schedule
+# lives in a published `/workflows` workflow (D6 — never a PM-app cron). This
+# function is the whole meeting point: the workflow's sweep node calls it with
+# nothing but the engine's identity, and everything it does is read from the
+# database at run time.
+
+
+def _months_before(midnight: datetime, months: int) -> datetime:
+    """``midnight`` moved back a whole number of calendar months.
+
+    Calendar months, not ``months * 30`` days, because the column is named in
+    months and an owner who says "one month" means "the 9th of last month",
+    not "30 days". The day clamps to the target month's length (May 31 minus
+    three months is Feb 28/29), which is the same choice every calendar
+    arithmetic library makes.
+    """
+    total = midnight.year * 12 + (midnight.month - 1) - months
+    year, month_index = divmod(total, 12)
+    month = month_index + 1
+    day = min(midnight.day, monthrange(year, month)[1])
+    return midnight.replace(year=year, month=month, day=day)
+
+
+def _project_midnight(moment: datetime, tz_name: str) -> datetime:
+    """The start of ``moment``'s day in the project's timezone.
+
+    This is what makes "a month untouched" defensible (P-28): the cutoff walks
+    back from the project's OWN midnight, so the boundary lands where the
+    people gardening the board live rather than at Greenwich. A bad zone name
+    (validated at write time, so this is belt-and-braces) falls back to UTC —
+    a sweep that skipped the project instead would silently switch the policy
+    off.
+    """
+    try:
+        zone = ZoneInfo(str(tz_name or "UTC"))
+    except Exception:
+        zone = ZoneInfo("UTC")
+    return datetime.combine(moment.astimezone(zone).date(), time.min, tzinfo=zone)
+
+
+def _closing_status(statuses: list[Any]) -> Any | None:
+    """The lane auto-close moves stale work to.
+
+    The project's first ``cancelled``-category lane (by position), else its
+    first ``done`` one. Cancelled preferred deliberately: a task nobody has
+    touched for months was abandoned, not finished, and calling it done would
+    inflate every completion report. There is no dedicated "default closing
+    status" flag in the schema (``is_default`` marks where NEW tasks land),
+    so the category ranking IS the model — stated here, tested, and easy to
+    replace with a flag if one ever earns its place.
+    """
+    for wanted in ("cancelled", "done"):
+        for row in statuses:  # already ordered by position
+            if str(row.category) == wanted:
+                return row
+    return None
+
+
+async def _sweep_candidates(
+    db: Any, root_id: str, status_ids: list[str], cutoff: datetime,
+) -> list[Any]:
+    """Unarchived tasks in these lanes, untouched since before ``cutoff``.
+
+    Membership is a POSITIVE ``ANY`` list on purpose — the caller hands in
+    exactly the lanes a pass may touch, so triage exemption and the archive
+    guard hold by construction rather than by a NOT somebody could drop.
+    """
+    if not status_ids:
+        return []
+    return (await db.execute(
+        text(
+            "SELECT * FROM pm_tasks "
+            "WHERE root_project_id = CAST(:root AS uuid) "
+            "AND archived_at IS NULL "
+            "AND status_id = ANY(CAST(:sids AS uuid[])) "
+            "AND updated_at < :cutoff"
+        ),
+        {"root": root_id, "sids": status_ids, "cutoff": cutoff},
+    )).fetchall()
+
+
+async def run_lifecycle_sweep(
+    db: Any, *, actor: str, now: datetime | None = None,
+) -> dict[str, Any]:
+    """Apply every root project's lifecycle policy once. Idempotent.
+
+    For each ROOT project with a policy enabled (both columns NULL — the
+    default — means the project is never touched):
+
+    * **archive** — tasks whose status category is done/cancelled and whose
+      ``updated_at`` is older than ``archive_after_months`` calendar months
+      before the project's own midnight leave the default surfaces
+      (``archived_at`` stamped, a ``system`` activity flagged automation).
+      Only closed-category lanes are eligible BY CONSTRUCTION, which is
+      WS-27w's manual archive guard restated as a query shape.
+    * **close** — open tasks (not closed, not triage — WS-27u's queue is a
+      place work WAITS) untouched beyond ``close_after_months`` move to the
+      project's closing status through :func:`core.apply_status_transition`,
+      so ``completed_at``, the ``status_change`` activity and recurrence all
+      behave exactly as if a person had moved the card.
+
+    A second run does nothing: archiving removes a task from the archive
+    pass's candidate set (``archived_at IS NULL``) and closing removes it
+    from the close pass's (its lane is now closed-category) — and the close
+    pass's transition bumps ``updated_at``, so even a task closed into a
+    project with a shorter archive window waits its full archive term.
+
+    Writes go through the ordinary service, as ``actor`` (the workflow's
+    ``system:workflow:<id>`` identity), so every change is an ordinary
+    timeline row wearing the automation flag.
+    """
+    moment = now or _clock()
+    roots = (await db.execute(
+        text("SELECT * FROM pm_projects WHERE parent_project_id IS NULL"),
+    )).fetchall()
+
+    swept = archived = closed = 0
+    for project in roots:
+        archive_months = getattr(project, "archive_after_months", None)
+        close_months = getattr(project, "close_after_months", None)
+        if archive_months is None and close_months is None:
+            continue
+        swept += 1
+        midnight = _project_midnight(
+            moment, getattr(project, "timezone", "UTC"),
+        )
+        statuses = (await db.execute(
+            text(
+                "SELECT * FROM pm_task_statuses "
+                "WHERE project_id = CAST(:pid AS uuid) ORDER BY position, name"
+            ),
+            {"pid": str(project.id)},
+        )).fetchall()
+
+        if archive_months is not None:
+            closed_ids = [
+                str(s.id) for s in statuses
+                if str(s.category) in CLOSING_CATEGORIES
+            ]
+            cutoff = _months_before(midnight, int(archive_months))
+            for task in await _sweep_candidates(
+                db, str(project.id), closed_ids, cutoff,
+            ):
+                await update_row(
+                    db, "pm_tasks", str(task.id), {"archived_at": moment},
+                )
+                await record_activity(
+                    db, activity_type="system", created_by=actor,
+                    task_id=str(task.id),
+                    body=(
+                        f"Task archived — untouched for "
+                        f"{int(archive_months)} month(s) after closing"
+                    ),
+                    automation=True,
+                )
+                archived += 1
+
+        if close_months is not None:
+            target = _closing_status(list(statuses))
+            open_ids = [
+                str(s.id) for s in statuses
+                if str(s.category) not in CLOSING_CATEGORIES
+                and str(s.category) != TRIAGE_CATEGORY
+            ]
+            if target is not None:
+                cutoff = _months_before(midnight, int(close_months))
+                for task in await _sweep_candidates(
+                    db, str(project.id), open_ids, cutoff,
+                ):
+                    await apply_status_transition(
+                        db, task, str(target.id), created_by=actor,
+                        automation=True,
+                    )
+                    closed += 1
+
+    return {
+        "projects": swept,
+        "archived": archived,
+        "closed": closed,
+        "skipped": not (archived or closed),
+    }