stack: Provider-Schicht für Aufgaben-Tracker mit Super-Productivity-Adapter (#124)
CI / verify (push) Successful in 48s
Release / release (push) Successful in 35s

Files changed:
- .gitignore
- CHANGES.md
- VERSION
- tools/CONTRACT.md
- tools/chemenu/commands/docs_verify.py
- tools/chemenu/commands/doctor.py
- tools/chemenu/config.py
- tools/chemenu/errors.py
- tools/chemenu/tasks/__init__.py
- tools/chemenu/tasks/config.py
- tools/chemenu/tasks/protocol.py
- tools/chemenu/tasks/superproductivity.py
- tools/chemenu/tests/test_doctor.py
- tools/chemenu/tests/test_superproductivity.py
- tools/chemenu/tests/test_tasks_config.py
- tools/chemenu/tests/test_tasks_protocol.py
This commit is contained in:
torben committed 2026-09-19 21:46:56 +02:00
1 parent ee24b6e5b8
commit 1875449b31
16 files changed
+1193 -6

No files matched your search

+3
View File
@@ -152,6 +152,9 @@ REQUIRED_IGNORE_CANARIES = (
# `.wikitool-telemetry.json` a few lines below - per-checkout, never
# committed.
".wikitool-upload.json",
# The task-tracker provider opt-in (Gitea #124) - same shape again:
# per-checkout, never committed, once a credential lands in it.
".wikitool-tasks.json",
)
REQUIRED_TRACKED_PATHS = (
"reports/CONTRACT.md",
+60 -3
View File
@@ -414,6 +414,61 @@ def check_upload_intake() -> Check:
)
def check_tasks_provider() -> Check:
"""Whether a task-tracker provider is configured for the GTD review
(Gitea #124), and whether it looks reachable.
Absent is `OK`, the same posture `check_upload_intake` takes on its own
config file: an instance with no tracker configured is legitimate, it
just cannot run the weekly review (#125) yet. A malformed config is a
`FAIL` for the same reason a malformed upload config is - it decides
which provider real credentials flow to, so a broken one must not read as
"nothing configured". Provider reachability itself never affects the
exit code, same as `check_git_repo`'s remote check: the app being closed
is normal, not a fault.
"""
from chemenu import config
from chemenu.errors import ValidationError
from chemenu.tasks import config as tasks_config
try:
cfg = tasks_config.read_config(config.ROOT)
except ValidationError as exc:
return Check(
"tasks-provider", "FAIL", str(exc),
f"Fix or delete {config.TASKS_CONFIG_FILENAME} - a broken one is not treated as "
"'no tracker configured'",
)
if cfg is None:
return Check(
"tasks-provider", "OK",
f"No {config.TASKS_CONFIG_FILENAME} - no task tracker configured (the weekly "
"review needs one, everything else does not)",
)
if cfg.provider == "superproductivity":
from chemenu.tasks import superproductivity as sp
try:
sp_cfg = sp.SuperProductivityConfig.from_dict(cfg.provider_config)
except ValidationError as exc:
return Check(
"tasks-provider", "FAIL", str(exc),
f"Fix the 'superproductivity' section of {config.TASKS_CONFIG_FILENAME}",
)
try:
snapshot_path = sp.latest_snapshot_path(sp_cfg)
read_state = f"read path OK, newest snapshot {rel_path(snapshot_path)}"
except ValidationError as exc:
read_state = f"read path not ready ({exc})"
api_state = "API reachable" if sp.health(sp_cfg) else "API not reachable (app not running?)"
return Check(
"tasks-provider", "OK", f"superproductivity: {read_state}; {api_state}",
)
return Check("tasks-provider", "OK", f"provider '{cfg.provider}' configured")
def check_session_id() -> Check:
"""Three-valued, not two: an explicit `WIKITOOL_SESSION_ID` and a
recognised harness variable (see `chemenu.session.HARNESS_ENV_VARS`) both
@@ -528,6 +583,7 @@ def run_doctor() -> list[Check]:
check_environment(),
check_publish_remotes(),
check_upload_intake(),
check_tasks_provider(),
check_generated_files(),
check_session_id(),
check_telemetry(),
@@ -540,9 +596,10 @@ def doctor_command(
):
"""Check that this instance is correctly configured: dependencies, author,
git identity/remote, published skills, structure, personalization, KB
conventions, generated files, session scoping, telemetry state, and
whether the MCP `submit` tool is armed. Read-only. Exits 1 only
if a check FAILs."""
conventions, generated files, session scoping, telemetry state, whether
the MCP `submit` tool is armed, and which task-tracker provider (if any)
is configured for the GTD review. Read-only. Exits 1 only if a check
FAILs."""
checks = run_doctor()
if json_out:
+9
View File
@@ -248,6 +248,15 @@ PUBLISH_REMOTES_FILENAME = ".wikitool-remotes.json"
# opt-in rather than a flag - see `chemenu.upload.read_config`.
UPLOAD_CONFIG_FILENAME = ".wikitool-upload.json"
# The task-tracker provider opt-in (Gitea #124, D25/D30): which provider this
# instance's GTD review reads/writes through, its connection details, and the
# review's three staleness thresholds. Same shape as the three files above -
# per-checkout, gitignored once a credential lands in it, no `.template` - and
# its absence is a legitimate state, the same posture `UPLOAD_CONFIG_FILENAME`
# takes: an instance with no tracker configured runs `doctor` and everything
# else just fine, it only can't run the weekly review (#125, not yet built).
TASKS_CONFIG_FILENAME = ".wikitool-tasks.json"
def default_author() -> str | None:
"""The author to stamp a new source page with, per instance.
+31
View File
@@ -29,3 +29,34 @@ class BackendError(ChemenuError, RuntimeError):
"""A dependency the core relies on was missing or failed - `rg` absent, a
search that had to be killed. Not the caller's argument, and not
necessarily permanent."""
class HumanInterventionRequired(ChemenuError):
"""A write this process cannot perform itself - not because the input was
wrong (that is `ValidationError`) and not because a dependency failed
(`BackendError`), but because the capability genuinely does not exist on
this side of the boundary. The canonical case (Gitea #124): Super
Productivity's local REST API has no project-creation endpoint, only
`GET /projects`, so `SuperProductivityWriter.create_project` cannot do the
one write `chemenu.tasks.protocol.TaskWriter` asks of it.
The CLI adapter (a future `wikitool new project`, Gitea #126) renders this
the same way it renders the four named gates in AGENTS.md's Gates section:
`commands._util.needs_clearance(str(exc))`, exit code 42 - "a human must
see the command's output before anything proceeds" applies here for the
same reason it applies to a mass update, just for a different cause. It is
not a fifth *named* gate (no threshold, no `--confirm` token to compute),
but the same exit code and the same posture: show the message verbatim,
stop, and do not improvise a workaround (AGENTS.md invariant 7).
`verify` is what makes this a request rather than a leap of faith: it is a
zero-argument callable that re-runs the read path and returns whether the
human's out-of-band step actually landed. A caller must invoke it after
the human confirms doing what `str(exc)` asked - "the user says they did
it" is never treated as "it happened" - and must refuse to proceed (and
ask again) while it still returns False.
"""
def __init__(self, message: str, *, verify):
super().__init__(message)
self.verify = verify
+12
View File
@@ -0,0 +1,12 @@
"""The task-tracker provider layer (Gitea #124, #119 D2/D3/D25).
`kb/` owns a project's durable memory; a task tracker owns its momentary open
loops (#119 D1). This package is the one place `wikitool` crosses that line -
never an instruction, never a second MCP server (#119 D25): a provider is a
Python object behind `chemenu.tasks.protocol.TaskReader`/`TaskWriter`, and
everything above this package (the future `wikitool review`/`new project`
commands, #125/#126) talks to that protocol and nothing provider-specific.
No command lives here yet - this package is a library, per #124's own scope
note ("Kein Kommando. Diese Schicht ist Bibliothek").
"""
+114
View File
@@ -0,0 +1,114 @@
"""Reads `.wikitool-tasks.json` (Gitea #124) - `config.TASKS_CONFIG_FILENAME`.
Same posture as `chemenu.upload.read_config`: absent means "no tracker
configured for this instance", a legitimate state that `doctor` reports as OK,
never as a fault. Malformed is a `ValidationError`, never silently ignored -
this file decides which provider (and which credentials) the review talks to,
so a broken one must not be read as "nothing configured".
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from chemenu import config
from chemenu.errors import ValidationError
# The providers this package ships an adapter for. Checked at read time so a
# typo in `provider` fails here, at the one place that knows the full list,
# rather than surfacing later as an unhelpful "unknown provider" from whatever
# code tried to dispatch on it.
KNOWN_PROVIDERS = ("superproductivity",)
@dataclass(frozen=True)
class Thresholds:
"""The weekly review's three staleness thresholds (#119 D10/D26) - each
named for the check it feeds, not for its unit alone, since two of the
three checks besides the "obvious" one also read a day count via
`datetime.timedelta`."""
stalled_waiting_days: int # check 2: a WAITING item older than this
unpaged_project_weeks: int # check 3: a tracker project with no kb/ page, older than this
someday_stale_months: int # check 5: a someday/maybe item untouched for this long
@dataclass(frozen=True)
class TasksConfig:
"""The opt-in, read from `.wikitool-tasks.json`."""
provider: str
thresholds: Thresholds
provider_config: dict[str, Any]
def read_config(root: "Any") -> "TasksConfig | None":
"""The tracker configuration for `root`, or `None` when the file is
absent - which means no tracker is configured for this instance, not that
one failed to load."""
from pathlib import Path
import json
path = Path(root) / config.TASKS_CONFIG_FILENAME
if not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ValidationError(
f"{config.TASKS_CONFIG_FILENAME} is unreadable ({exc}). It decides which task "
"tracker the weekly review talks to, so a broken file is not treated as 'no "
"tracker configured' - fix it or delete it deliberately."
) from exc
if not isinstance(data, dict):
raise ValidationError(f"{config.TASKS_CONFIG_FILENAME} must contain a JSON object.")
expected = (
'{"schema": 1, "provider": "superproductivity", '
'"thresholds": {"stalled_waiting_days": 14, "unpaged_project_weeks": 3, '
'"someday_stale_months": 5}, "superproductivity": {"backups_dir": "...", '
'"api_base_url": "http://127.0.0.1:3876", "api_token": "..."}}'
)
try:
provider = str(data["provider"])
raw_thresholds = data["thresholds"]
stalled_waiting_days = int(raw_thresholds["stalled_waiting_days"])
unpaged_project_weeks = int(raw_thresholds["unpaged_project_weeks"])
someday_stale_months = int(raw_thresholds["someday_stale_months"])
except (KeyError, TypeError, ValueError) as exc:
raise ValidationError(
f"{config.TASKS_CONFIG_FILENAME} is missing or misshapes a required field ({exc}). "
f"Expected: {expected}"
) from exc
if provider not in KNOWN_PROVIDERS:
raise ValidationError(
f"{config.TASKS_CONFIG_FILENAME}: unknown provider {provider!r}. "
f"Known: {', '.join(KNOWN_PROVIDERS)}"
)
for field_name, value in (
("stalled_waiting_days", stalled_waiting_days),
("unpaged_project_weeks", unpaged_project_weeks),
("someday_stale_months", someday_stale_months),
):
if value <= 0:
raise ValidationError(
f"{config.TASKS_CONFIG_FILENAME}: thresholds.{field_name} must be positive."
)
provider_config = data.get(provider)
if not isinstance(provider_config, dict):
raise ValidationError(
f"{config.TASKS_CONFIG_FILENAME}: missing or non-object {provider!r} section "
f"holding that provider's own connection settings. Expected: {expected}"
)
return TasksConfig(
provider=provider,
thresholds=Thresholds(
stalled_waiting_days=stalled_waiting_days,
unpaged_project_weeks=unpaged_project_weeks,
someday_stale_months=someday_stale_months,
),
provider_config=provider_config,
)
+133
View File
@@ -0,0 +1,133 @@
"""The provider-agnostic read/write shape (Gitea #124, #119 D26/D31).
Every provider adapter under `chemenu.tasks` implements `TaskReader` and, if
it can, `TaskWriter` - two separate `Protocol`s rather than one, because #124's
own acceptance criteria requires exactly that: "ein Adapter kann den
Schreibpfad nicht anbieten, ohne dass der Lesepfad davon beruehrt wird". A
provider whose write path cannot exist (see `SuperProductivityWriter`) simply
does not implement `TaskWriter` - nothing here forces it to.
The read shape is fixed by what the weekly review (#119 D26, built in #125)
needs and nothing more: which projects exist and when they were created, how
many open items each one has and which of those are `WAITING` with a
`follow_up_at` (#119 D9/D30 - the *only* two machine-readable parts of a
waiting-for item; the person stays in the title's free text), and which
someday/maybe items exist and when they last moved. None of this is cached
here - a `TaskReader` re-reads on every call, so a caller checking `verify()`
after a human's out-of-band step (see `chemenu.errors.HumanInterventionRequired`)
sees the current state, not a snapshot from before that step.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from typing import Optional, Protocol, Sequence
@dataclass(frozen=True)
class ProjectSummary:
"""One tracker project, as the review needs it: its name (the sole join
key with a `kb/gtd/` page, #119 D8) and when it was created."""
name: str
created: Optional[date]
@dataclass(frozen=True)
class WaitingItem:
"""One open item carrying the `WAITING` status (#119 D9/D30).
`title` is shown verbatim, person and all - the review never parses it.
`follow_up_at` is the one machine-readable date, and it is deliberately
**not** the item's due date (#119 D9: "ausdruecklich nicht das
Faelligkeitsdatum") - a provider that has no separate concept for this
must not fall back to reusing the due date, it must decide it cannot
supply the field and leave it `None` instead.
"""
title: str
follow_up_at: Optional[date]
@dataclass(frozen=True)
class OpenItems:
"""A project's momentary open-loop count, plus the subset that is
`WAITING`. `count` includes the waiting items - it is "how many open
items", not "how many open items that aren't waiting"."""
count: int
waiting: Sequence[WaitingItem]
@dataclass(frozen=True)
class SomedayItem:
"""One someday/maybe item: its title and when it last changed, for
check 5's staleness read (#119 D26)."""
title: str
modified: Optional[date]
def normalize_project_name(name: str) -> str:
"""The case- and whitespace-normalized form of a project name (#119 D8):
collapse internal whitespace, then casefold. Used both ways round - to
preflight a name against the read path before a create, and to check
whether a human's out-of-band creation (`HumanInterventionRequired.verify`)
actually landed - so the same normalization must decide both, or a name
that passes one check could fail the other."""
return " ".join(name.strip().split()).casefold()
class TaskReader(Protocol):
"""The read path every provider adapter must implement."""
def projects(self) -> list[ProjectSummary]:
"""Every project the tracker currently knows, in no particular
order."""
...
def open_items(self, project_name: str) -> OpenItems:
"""Open items for the project named `project_name` (matched
case-normalized, #119 D8). A project the tracker does not know
returns `OpenItems(count=0, waiting=())` - "no open items" and "no
such project" are not distinguished here, because check 3 (#119 D26)
is what tells those apart, over the read path's `projects()` list."""
...
def someday_items(self) -> list[SomedayItem]:
"""Every someday/maybe item the tracker currently holds, across all
projects."""
...
class TaskWriter(Protocol):
"""The write path a provider adapter offers only if it actually can
(#124's own acceptance criteria: offering this must not touch the read
path's availability)."""
def create_project(self, name: str) -> None:
"""Create a tracker project named `name`, after checking `name` is
not already taken (case-normalized, #119 D8) via the read path.
Raises `chemenu.errors.ValidationError` if the name collides, or if
the provider is reachable but refuses for a reason a human cannot fix
by way of `chemenu.errors.HumanInterventionRequired` (e.g. the
provider app is simply not running). Raises
`chemenu.errors.HumanInterventionRequired` if this provider has no way
to create a project itself and a human must do it out of band - see
that class's docstring for the full contract, including `verify()`.
"""
...
def find_project(reader: TaskReader, name: str) -> Optional[ProjectSummary]:
"""The project matching `name` case-normalized (#119 D8), or `None`.
Shared by a `TaskWriter`'s preflight collision check and by a
`HumanInterventionRequired.verify()` closure - both are the same
question, "does a project by this name exist right now", asked at two
different moments."""
target = normalize_project_name(name)
for project in reader.projects():
if normalize_project_name(project.name) == target:
return project
return None
+318
View File
@@ -0,0 +1,318 @@
"""The Super Productivity adapter (Gitea #124, #119 D2/D8/D9/D30).
Read and write deliberately use different transports (#124's own design note,
carried as its first acceptance criterion): the read path is headless file
access, the write path is Super Productivity's own local REST API - so a
review can run with the app closed, and a create only needs the app open when
it actually has to ask it for something.
## Read path: the backup snapshot, not a live `db.json`
Desktop Super Productivity keeps its live state in IndexedDB, not in a flat
file called `db.json` on disk - there is no such file to read headlessly.
What *does* exist as a plain file is a periodic snapshot: `electron/backup.ts`
writes the complete app state as `JSON.stringify(data)` into
`<userData>/backups/<timestamp>.json` on every backup, newest-timestamp-last
by filename (its own comment: "timestamps sort lexically"). That snapshot's
top-level shape is `AppDataComplete`/`AppDataCompleteLegacy`
(`src/app/op-log/model/model-config.ts` / `src/app/imex/sync/sync.model.ts`),
keyed by feature name - `task`, `project`, `tag`, ... - and this reader only
looks at the three keys it needs, each an `@ngrx/entity` `EntityState`
(`{"ids": [...], "entities": {...}}`, `packages/plugin-api/src/types.ts`
`Task`/`Project`/`Tag`). Verified against the `master` branch of
`super-productivity/super-productivity` on 2026-09-19 - unversioned, per
#124's own note, so a future release is free to reshape it without warning,
which is exactly why every read below fails loudly on a shape it does not
recognize rather than guessing.
## The two GTD conventions this adapter encodes (#119 D9/D30)
Only the `WAITING` status and `follow_up_at` are machine-readable, and Super
Productivity has no native field for either:
- **`WAITING`** is a tag named `waiting` (case-insensitively), attached to the
task. Any other tag is left alone.
- **`follow_up_at`** is the task's own `remindAt` (a reminder timestamp) -
deliberately not `dueDay`/`dueWithTime` (#119 D9: "ausdruecklich nicht das
Faelligkeitsdatum"). A task with no reminder set has no `follow_up_at`, full
stop; this adapter never substitutes the due date for it.
## Someday/Maybe: a project's own backlog
#124 (later, #119) left this as "zu verifizieren": Super Productivity's
`ProjectBasicCfg.backlogTaskIds` is exactly this - a second, separate list of
task ids per project, apart from the active `taskIds` list a project's board
shows. This adapter reads someday/maybe items from `backlogTaskIds`, one
project at a time; no tag convention is needed.
## Write path: no project-creation endpoint exists
The local REST API (`electron/local-rest-api-handler.service.ts`, verified the
same day) routes `GET /projects` but has no `POST /projects` at all - task
CRUD exists, project CRUD does not. `SuperProductivityWriter.create_project`
can therefore not create a project itself; see its docstring and
`chemenu.errors.HumanInterventionRequired`.
"""
from __future__ import annotations
import json
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any, Optional
from chemenu.errors import HumanInterventionRequired, ValidationError
from chemenu.tasks.protocol import (
OpenItems,
ProjectSummary,
SomedayItem,
WaitingItem,
find_project,
normalize_project_name,
)
# The tag title that means "WAITING" (#119 D9/D30), matched case-insensitively
# - this instance's own convention, not something Super Productivity defines.
WAITING_TAG_TITLE = "waiting"
DEFAULT_API_BASE_URL = "http://127.0.0.1:3876"
_EXPECTED_PROVIDER_CONFIG = (
'{"backups_dir": "~/.config/superProductivity/backups", '
'"api_base_url": "http://127.0.0.1:3876", "api_token": "..."}'
' (or "db_path" instead of "backups_dir" to pin one exact file)'
)
@dataclass(frozen=True)
class SuperProductivityConfig:
"""This provider's own section of `.wikitool-tasks.json`
(`TasksConfig.provider_config`)."""
# Exactly one of these two names where to read from. `db_path` wins when
# both are set - it names one exact file, which is a stronger statement
# than "the newest file in this directory".
backups_dir: Optional[Path]
db_path: Optional[Path]
api_base_url: str
# Unused by this module today - nothing it calls needs authentication
# (`GET /health` is the one unauthenticated exception, and there is no
# write call at all, see the module docstring). Carried through anyway so
# a future capability that does need it does not require a config-shape
# migration to add it.
api_token: Optional[str]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "SuperProductivityConfig":
if not isinstance(data, dict):
raise ValidationError(
f"superproductivity config must be an object. Expected: {_EXPECTED_PROVIDER_CONFIG}"
)
backups_dir = data.get("backups_dir")
db_path = data.get("db_path")
if backups_dir is None and db_path is None:
raise ValidationError(
"superproductivity config needs 'backups_dir' or 'db_path' - the read path has "
f"nothing to read otherwise. Expected: {_EXPECTED_PROVIDER_CONFIG}"
)
api_base_url = str(data.get("api_base_url") or DEFAULT_API_BASE_URL)
api_token = data.get("api_token")
if api_token is not None and not isinstance(api_token, str):
raise ValidationError("superproductivity config: api_token must be a string.")
return cls(
backups_dir=Path(backups_dir).expanduser() if backups_dir else None,
db_path=Path(db_path).expanduser() if db_path else None,
api_base_url=api_base_url,
api_token=api_token,
)
def latest_snapshot_path(cfg: SuperProductivityConfig) -> Path:
"""The one file to read: `db_path` if given, else the lexically-greatest
`*.json` filename under `backups_dir` - the same ordering
`electron/backup.ts` itself relies on ("timestamps sort lexically")."""
if cfg.db_path is not None:
if not cfg.db_path.is_file():
raise ValidationError(
f"superproductivity: db_path does not exist: {cfg.db_path}"
)
return cfg.db_path
directory = cfg.backups_dir
assert directory is not None # from_dict guarantees at least one is set
if not directory.is_dir():
raise ValidationError(
f"superproductivity: backups_dir does not exist: {directory}"
)
candidates = sorted(directory.glob("*.json"))
if not candidates:
raise ValidationError(
f"superproductivity: no *.json backup file found under {directory}. Take a "
"backup from Super Productivity (Settings -> Backup & Sync -> Local backups), "
"or point db_path/backups_dir at where it actually writes them."
)
return candidates[-1]
def _load_snapshot(cfg: SuperProductivityConfig) -> dict[str, Any]:
path = latest_snapshot_path(cfg)
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ValidationError(f"superproductivity: cannot read {path} ({exc}).") from exc
if not isinstance(data, dict):
raise ValidationError(
f"superproductivity: {path} does not contain a JSON object at its top level."
)
return data
def _entity_state(snapshot: dict[str, Any], key: str, source: Path) -> dict[str, dict]:
"""`snapshot[key]` as an `@ngrx/entity` `{"ids": [...], "entities": {...}}`
map, or a loud `ValidationError` naming exactly what was expected - Super
Productivity's internal model is unversioned (module docstring), so a
shape drift here is expected eventually, not a bug in this adapter."""
value = snapshot.get(key)
if (
not isinstance(value, dict)
or not isinstance(value.get("ids"), list)
or not isinstance(value.get("entities"), dict)
):
raise ValidationError(
f"superproductivity: {source} has no usable '{key}' entity state "
f"({{\"ids\": [...], \"entities\": {{...}}}}). Its internal shape is unversioned and "
"this file may be from a Super Productivity version this adapter does not know - "
f"found: {type(value).__name__ if value is not None else 'missing'}."
)
return value["entities"]
def _epoch_ms_to_date(value: Any) -> Optional[date]:
if not isinstance(value, (int, float)):
return None
return datetime.fromtimestamp(value / 1000, tz=timezone.utc).date()
class SuperProductivityReader:
"""`TaskReader` over a Super Productivity backup snapshot. Re-reads the
snapshot on every call - see `chemenu.errors.HumanInterventionRequired`
for why that matters."""
def __init__(self, cfg: SuperProductivityConfig):
self._cfg = cfg
def _read(self) -> tuple[Path, dict[str, dict], dict[str, dict], dict[str, dict]]:
path = latest_snapshot_path(self._cfg)
snapshot = _load_snapshot(self._cfg)
projects = _entity_state(snapshot, "project", path)
tasks = _entity_state(snapshot, "task", path)
tags = _entity_state(snapshot, "tag", path)
return path, projects, tasks, tags
def projects(self) -> list[ProjectSummary]:
_, projects, _, _ = self._read()
return [
ProjectSummary(
name=str(record.get("title", "")),
created=_epoch_ms_to_date(record.get("created")),
)
for record in projects.values()
]
def _is_waiting(self, task: dict, tags: dict[str, dict]) -> bool:
for tag_id in task.get("tagIds") or []:
tag = tags.get(tag_id)
if tag and str(tag.get("title", "")).strip().casefold() == WAITING_TAG_TITLE:
return True
return False
def open_items(self, project_name: str) -> OpenItems:
_, projects, tasks, tags = self._read()
target = normalize_project_name(project_name)
project = next(
(p for p in projects.values() if normalize_project_name(str(p.get("title", ""))) == target),
None,
)
if project is None:
return OpenItems(count=0, waiting=())
waiting: list[WaitingItem] = []
count = 0
for task_id in project.get("taskIds") or []:
task = tasks.get(task_id)
if task is None or task.get("isDone"):
continue
count += 1
if self._is_waiting(task, tags):
waiting.append(
WaitingItem(
title=str(task.get("title", "")),
follow_up_at=_epoch_ms_to_date(task.get("remindAt")),
)
)
return OpenItems(count=count, waiting=tuple(waiting))
def someday_items(self) -> list[SomedayItem]:
_, projects, tasks, _ = self._read()
items: list[SomedayItem] = []
for project in projects.values():
for task_id in project.get("backlogTaskIds") or []:
task = tasks.get(task_id)
if task is None or task.get("isDone"):
continue
items.append(
SomedayItem(
title=str(task.get("title", "")),
modified=_epoch_ms_to_date(task.get("updated") or task.get("created")),
)
)
return items
def health(cfg: SuperProductivityConfig, *, timeout: float = 2.0) -> bool:
"""Whether the local REST API answers `GET /health` right now - the one
unauthenticated endpoint (module docstring). Never raises: an unreachable
app is an ordinary, expected state (`doctor` reports it, it does not
FAIL), not a defect in this adapter."""
url = cfg.api_base_url.rstrip("/") + "/health"
try:
with urllib.request.urlopen(url, timeout=timeout) as response: # noqa: S310 - localhost only
return 200 <= response.status < 300
except (urllib.error.URLError, OSError, ValueError):
return False
class SuperProductivityWriter:
"""`TaskWriter` over the local REST API - except there is no API call
this can actually make, see the module docstring."""
def __init__(self, cfg: SuperProductivityConfig, reader: SuperProductivityReader):
self._cfg = cfg
self._reader = reader
def create_project(self, name: str) -> None:
"""Never creates anything. Preflights the name against the read path
(#119 D8) and, if it is free, raises `HumanInterventionRequired`
naming the one thing a human must do - Super Productivity's local
REST API has no project-creation endpoint at all (module docstring),
so this is not a missing feature in this adapter, it is a missing
endpoint upstream."""
existing = find_project(self._reader, name)
if existing is not None:
raise ValidationError(
f"A project named '{name}' (case-insensitively) already exists in "
"Super Productivity - nothing was created."
)
def _verify() -> bool:
return find_project(self._reader, name) is not None
raise HumanInterventionRequired(
"Super Productivity's local REST API has no project-creation endpoint "
"(only GET /projects) - this cannot be automated.\n"
f" 1. Open Super Productivity.\n"
f" 2. Create a project named exactly: {name}\n"
" 3. Tell the agent you have done this, so it can re-check and continue.",
verify=_verify,
)
+44
View File
@@ -3,6 +3,7 @@ never FAIL, and each check independently reports FAIL when its precondition
is missing."""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
@@ -346,6 +347,49 @@ def test_upload_intake_fails_on_a_malformed_config(instance):
assert _status(checks, "upload-intake") == "FAIL"
def test_tasks_provider_ok_when_no_config_file(instance):
checks = doctor.run_doctor()
assert _status(checks, "tasks-provider") == "OK"
assert "no task tracker configured" in _detail(checks, "tasks-provider")
def test_tasks_provider_fails_on_a_malformed_config(instance):
(config.ROOT / config.TASKS_CONFIG_FILENAME).write_text("{not json", encoding="utf-8")
checks = doctor.run_doctor()
assert _status(checks, "tasks-provider") == "FAIL"
def test_tasks_provider_fails_on_a_bad_superproductivity_section(instance):
(config.ROOT / config.TASKS_CONFIG_FILENAME).write_text(
'{"schema": 1, "provider": "superproductivity", "thresholds": '
'{"stalled_waiting_days": 14, "unpaged_project_weeks": 3, "someday_stale_months": 5}, '
'"superproductivity": {}}',
encoding="utf-8",
)
checks = doctor.run_doctor()
assert _status(checks, "tasks-provider") == "FAIL"
def test_tasks_provider_ok_but_names_the_unready_read_path_when_configured(instance):
backups = config.ROOT / "sp-backups"
(config.ROOT / config.TASKS_CONFIG_FILENAME).write_text(
json.dumps({
"schema": 1, "provider": "superproductivity",
"thresholds": {
"stalled_waiting_days": 14, "unpaged_project_weeks": 3,
"someday_stale_months": 5,
},
"superproductivity": {"backups_dir": str(backups)},
}),
encoding="utf-8",
)
checks = doctor.run_doctor()
assert _status(checks, "tasks-provider") == "OK"
detail = _detail(checks, "tasks-provider")
assert "not ready" in detail
assert "not reachable" in detail
def test_missing_generated_file_fails(instance):
config.LOG_FILE.unlink()
checks = doctor.run_doctor()
@@ -0,0 +1,298 @@
"""Tests for `chemenu.tasks.superproductivity` (Gitea #124).
The fixture snapshot below mirrors the real shape verified against
`super-productivity/super-productivity`'s `master` branch on 2026-09-19: a
flat top-level object with `project`/`task`/`tag` as `@ngrx/entity`
`{"ids": [...], "entities": {...}}` maps (see the module docstring for the
exact source files). No test here starts a real Super Productivity instance,
touches the network beyond a loopback socket, or reads a path under the
developer's home - everything is built inside `tmp_path`
(`instructions/dev/testing-conventions.md`).
"""
from __future__ import annotations
import http.server
import json
import threading
from datetime import date, datetime, timezone
from pathlib import Path
import pytest
from chemenu.errors import HumanInterventionRequired, ValidationError
from chemenu.tasks import superproductivity as sp
def _ms(year: int, month: int, day: int) -> int:
return int(datetime(year, month, day, tzinfo=timezone.utc).timestamp() * 1000)
def _entity_state(records: dict[str, dict]) -> dict:
return {"ids": list(records.keys()), "entities": records}
def _snapshot() -> dict:
projects = {
"p1": {
"id": "p1",
"title": "Ship Chemenu 7.0",
"created": _ms(2026, 1, 1),
"taskIds": ["t1", "t2"],
"backlogTaskIds": ["t3"],
},
"p2": {
"id": "p2",
"title": "Kueche renovieren",
"created": _ms(2026, 2, 1),
"taskIds": ["t4"],
"backlogTaskIds": [],
},
}
tasks = {
"t1": {
"id": "t1",
"title": "Warte auf Angebot vom Elektriker - Tobias",
"projectId": "p1",
"isDone": False,
"tagIds": ["tag-wait"],
"remindAt": _ms(2026, 3, 1),
},
"t2": {
"id": "t2",
"title": "Kickoff-Meeting vorbereiten",
"projectId": "p1",
"isDone": False,
"tagIds": [],
},
"t3": {
"id": "t3",
"title": "Irgendwann Keller aufraeumen",
"isDone": False,
"tagIds": [],
"updated": _ms(2026, 1, 15),
},
"t4": {
"id": "t4",
"title": "Angebot einholen",
"projectId": "p2",
"isDone": True,
"tagIds": [],
},
}
tags = {
"tag-wait": {"id": "tag-wait", "title": "Waiting"},
"tag-urgent": {"id": "tag-urgent", "title": "Urgent"},
}
return {
"project": _entity_state(projects),
"task": _entity_state(tasks),
"tag": _entity_state(tags),
}
def _write_snapshot(path: Path, data: dict | None = None) -> None:
path.write_text(json.dumps(data if data is not None else _snapshot()), encoding="utf-8")
@pytest.fixture
def cfg(tmp_path) -> sp.SuperProductivityConfig:
db_path = tmp_path / "db.json"
_write_snapshot(db_path)
return sp.SuperProductivityConfig(
backups_dir=None, db_path=db_path, api_base_url=sp.DEFAULT_API_BASE_URL, api_token=None
)
# --- from_dict ---------------------------------------------------------------
def test_from_dict_requires_backups_dir_or_db_path():
with pytest.raises(ValidationError):
sp.SuperProductivityConfig.from_dict({"api_base_url": "http://127.0.0.1:3876"})
def test_from_dict_defaults_the_api_base_url():
cfg_ = sp.SuperProductivityConfig.from_dict({"db_path": "/x/db.json"})
assert cfg_.api_base_url == sp.DEFAULT_API_BASE_URL
def test_from_dict_expands_user_in_paths(monkeypatch):
monkeypatch.setenv("HOME", "/home/fixture")
cfg_ = sp.SuperProductivityConfig.from_dict({"backups_dir": "~/backups"})
assert cfg_.backups_dir == Path("/home/fixture/backups")
# --- latest_snapshot_path / schema drift -------------------------------------
def test_db_path_must_exist(tmp_path):
cfg_ = sp.SuperProductivityConfig(
backups_dir=None, db_path=tmp_path / "missing.json",
api_base_url=sp.DEFAULT_API_BASE_URL, api_token=None,
)
with pytest.raises(ValidationError):
sp.latest_snapshot_path(cfg_)
def test_backups_dir_must_exist(tmp_path):
cfg_ = sp.SuperProductivityConfig(
backups_dir=tmp_path / "nope", db_path=None,
api_base_url=sp.DEFAULT_API_BASE_URL, api_token=None,
)
with pytest.raises(ValidationError):
sp.latest_snapshot_path(cfg_)
def test_backups_dir_with_no_json_files_is_an_error(tmp_path):
backups = tmp_path / "backups"
backups.mkdir()
cfg_ = sp.SuperProductivityConfig(
backups_dir=backups, db_path=None, api_base_url=sp.DEFAULT_API_BASE_URL, api_token=None,
)
with pytest.raises(ValidationError):
sp.latest_snapshot_path(cfg_)
def test_backups_dir_picks_the_lexically_latest_file(tmp_path):
backups = tmp_path / "backups"
backups.mkdir()
older = _snapshot()
older["project"]["entities"]["p1"]["title"] = "Old Snapshot Project"
newer = _snapshot()
_write_snapshot(backups / "2026-01-01T000000Z.json", older)
_write_snapshot(backups / "2026-02-01T000000Z.json", newer)
cfg_ = sp.SuperProductivityConfig(
backups_dir=backups, db_path=None, api_base_url=sp.DEFAULT_API_BASE_URL, api_token=None,
)
reader = sp.SuperProductivityReader(cfg_)
names = {p.name for p in reader.projects()}
assert "Ship Chemenu 7.0" in names
assert "Old Snapshot Project" not in names
def test_schema_drift_on_task_key_fails_loud_not_silent(cfg):
data = _snapshot()
data["task"] = ["not", "an", "entity", "state"]
_write_snapshot(cfg.db_path, data)
reader = sp.SuperProductivityReader(cfg)
with pytest.raises(ValidationError, match="task"):
reader.projects()
def test_missing_top_level_key_fails_loud(cfg):
data = _snapshot()
del data["project"]
_write_snapshot(cfg.db_path, data)
reader = sp.SuperProductivityReader(cfg)
with pytest.raises(ValidationError, match="project"):
reader.projects()
# --- read path ----------------------------------------------------------------
def test_projects_lists_name_and_created(cfg):
reader = sp.SuperProductivityReader(cfg)
by_name = {p.name: p for p in reader.projects()}
assert by_name["Ship Chemenu 7.0"].created == date(2026, 1, 1)
assert by_name["Kueche renovieren"].created == date(2026, 2, 1)
def test_open_items_counts_undone_tasks_and_matches_name_case_insensitively(cfg):
reader = sp.SuperProductivityReader(cfg)
result = reader.open_items("ship CHEMENU 7.0")
assert result.count == 2
def test_open_items_reports_waiting_with_follow_up_at_not_due_date(cfg):
reader = sp.SuperProductivityReader(cfg)
result = reader.open_items("Ship Chemenu 7.0")
assert len(result.waiting) == 1
waiting = result.waiting[0]
assert waiting.title == "Warte auf Angebot vom Elektriker - Tobias"
assert waiting.follow_up_at == date(2026, 3, 1)
def test_open_items_excludes_done_tasks(cfg):
reader = sp.SuperProductivityReader(cfg)
assert reader.open_items("Kueche renovieren").count == 0
def test_open_items_unknown_project_is_empty_not_an_error(cfg):
reader = sp.SuperProductivityReader(cfg)
result = reader.open_items("No Such Project")
assert result.count == 0
assert result.waiting == ()
def test_someday_items_come_from_backlog_task_ids_only(cfg):
reader = sp.SuperProductivityReader(cfg)
items = reader.someday_items()
assert len(items) == 1
assert items[0].title == "Irgendwann Keller aufraeumen"
assert items[0].modified == date(2026, 1, 15)
# --- write path: HumanInterventionRequired -----------------------------------
def test_create_project_refuses_a_name_collision(cfg):
reader = sp.SuperProductivityReader(cfg)
writer = sp.SuperProductivityWriter(cfg, reader)
with pytest.raises(ValidationError):
writer.create_project("ship chemenu 7.0")
def test_create_project_asks_a_human_and_verify_reflects_the_read_path(cfg):
reader = sp.SuperProductivityReader(cfg)
writer = sp.SuperProductivityWriter(cfg, reader)
with pytest.raises(HumanInterventionRequired) as excinfo:
writer.create_project("Kueche renovieren, Phase 2")
exc = excinfo.value
assert "POST" not in str(exc) # instructions are for a human, not an HTTP client
assert "Kueche renovieren, Phase 2" in str(exc)
assert exc.verify() is False
data = _snapshot()
data["project"]["ids"].append("p3")
data["project"]["entities"]["p3"] = {
"id": "p3", "title": "Kueche renovieren, Phase 2", "created": _ms(2026, 4, 1),
"taskIds": [], "backlogTaskIds": [],
}
_write_snapshot(cfg.db_path, data)
assert exc.verify() is True
# --- health -------------------------------------------------------------------
def test_health_is_false_when_nothing_listens(tmp_path):
cfg_ = sp.SuperProductivityConfig(
backups_dir=None, db_path=tmp_path / "db.json",
api_base_url="http://127.0.0.1:1", api_token=None,
)
assert sp.health(cfg_, timeout=0.5) is False
class _HealthHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self): # noqa: N802 - stdlib method name
self.send_response(200)
self.end_headers()
self.wfile.write(b'{"ok": true}')
def log_message(self, *args): # silence stderr noise during the test run
pass
def test_health_is_true_when_the_endpoint_answers(tmp_path):
server = http.server.HTTPServer(("127.0.0.1", 0), _HealthHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
port = server.server_address[1]
cfg_ = sp.SuperProductivityConfig(
backups_dir=None, db_path=tmp_path / "db.json",
api_base_url=f"http://127.0.0.1:{port}", api_token=None,
)
assert sp.health(cfg_, timeout=2.0) is True
finally:
server.shutdown()
thread.join(timeout=2)
+81
View File
@@ -0,0 +1,81 @@
"""Tests for `chemenu.tasks.config.read_config` (Gitea #124) - same posture as
`chemenu.upload.read_config`: absent is `None` ("no tracker configured"),
malformed is a `ValidationError`, never silently "unrestricted"."""
from __future__ import annotations
import json
import pytest
from chemenu.errors import ValidationError
from chemenu.tasks import config as tasks_config
VALID = {
"schema": 1,
"provider": "superproductivity",
"thresholds": {
"stalled_waiting_days": 14,
"unpaged_project_weeks": 3,
"someday_stale_months": 5,
},
"superproductivity": {
"backups_dir": "/tmp/does-not-need-to-exist-for-parsing",
"api_base_url": "http://127.0.0.1:3876",
},
}
def _write(root, data):
(root / ".wikitool-tasks.json").write_text(json.dumps(data), encoding="utf-8")
def test_absent_is_none(tmp_path):
assert tasks_config.read_config(tmp_path) is None
def test_malformed_json_is_an_error_not_unconfigured(tmp_path):
(tmp_path / ".wikitool-tasks.json").write_text("{not json", encoding="utf-8")
with pytest.raises(ValidationError):
tasks_config.read_config(tmp_path)
def test_a_valid_config_parses(tmp_path):
_write(tmp_path, VALID)
cfg = tasks_config.read_config(tmp_path)
assert cfg.provider == "superproductivity"
assert cfg.thresholds.stalled_waiting_days == 14
assert cfg.thresholds.unpaged_project_weeks == 3
assert cfg.thresholds.someday_stale_months == 5
assert cfg.provider_config["backups_dir"] == "/tmp/does-not-need-to-exist-for-parsing"
def test_unknown_provider_is_rejected(tmp_path):
data = json.loads(json.dumps(VALID))
data["provider"] = "asana"
_write(tmp_path, data)
with pytest.raises(ValidationError):
tasks_config.read_config(tmp_path)
def test_missing_threshold_field_is_rejected(tmp_path):
data = json.loads(json.dumps(VALID))
del data["thresholds"]["someday_stale_months"]
_write(tmp_path, data)
with pytest.raises(ValidationError):
tasks_config.read_config(tmp_path)
def test_non_positive_threshold_is_rejected(tmp_path):
data = json.loads(json.dumps(VALID))
data["thresholds"]["stalled_waiting_days"] = 0
_write(tmp_path, data)
with pytest.raises(ValidationError):
tasks_config.read_config(tmp_path)
def test_missing_provider_section_is_rejected(tmp_path):
data = json.loads(json.dumps(VALID))
del data["superproductivity"]
_write(tmp_path, data)
with pytest.raises(ValidationError):
tasks_config.read_config(tmp_path)
@@ -0,0 +1,44 @@
"""Tests for `chemenu.tasks.protocol` (Gitea #124) - the provider-agnostic
read/write shape and its one piece of real logic, name normalization."""
from __future__ import annotations
from datetime import date
from chemenu.tasks.protocol import (
ProjectSummary,
find_project,
normalize_project_name,
)
def test_normalize_collapses_whitespace_and_case():
assert normalize_project_name(" Ship Chemenu 7.0 ") == "ship chemenu 7.0"
assert normalize_project_name("Ship Chemenu 7.0") == normalize_project_name(
"ship chemenu 7.0"
)
class _Reader:
def __init__(self, projects):
self._projects = projects
def projects(self):
return self._projects
def open_items(self, project_name): # pragma: no cover - not exercised here
raise NotImplementedError
def someday_items(self): # pragma: no cover - not exercised here
raise NotImplementedError
def test_find_project_matches_case_and_whitespace_insensitively():
reader = _Reader([ProjectSummary(name="Ship Chemenu 7.0", created=date(2026, 1, 1))])
found = find_project(reader, "ship chemenu 7.0")
assert found is not None
assert found.name == "Ship Chemenu 7.0"
def test_find_project_returns_none_when_absent():
reader = _Reader([ProjectSummary(name="Other", created=None)])
assert find_project(reader, "Ship Chemenu 7.0") is None