chemenu/tools/chemenu/config_shapes.py

"""The shapes of the tiered settings sections.

`wikitool.json` and `.wikitool-local.json` (`chemenu.settings`) hold one section
per feature; each section has exactly one record here, and that record is the
only place its shape is written down. The loader that reads a section takes its
field names, required fields, defaults, coercions and the tier of each field
from the record; `docs reference --apply` renders the same record into
`docs/reference/configuration.md`. A second description of a shape, in a loader
or in prose, is what used to drift.

Every field declares the tier it belongs to (`INSTANCE`, `CHECKOUT`, `SECRET`),
and the rule that a field naming a program or affecting a gate is never
`INSTANCE` is checked when the `Field` is declared (`Field.__post_init__`).

The records describe what the loaders **accept**, leniency included: `int()` on
a number given as a string, `bool()` on `enabled`. `check()` never raises. Each
loader keeps its own posture towards a problem - `publish` stops through
`fail()`, the upload and tracker loaders raise `ValidationError`, and telemetry
drops what it cannot use in silence, because a hook handler reads it on every
tool call. `deviations()` is the strict reading `doctor` uses to report what
that silence would otherwise hide.

The three machine-written files beside them - `.wikitool-release.json`,
`.wikitool-kb.json`, `.wikitool-tools.json` - are deliberately not here: nobody
writes them by hand, so they have no shape for an operator to look up.

Stdlib only, and importing nothing from this package but `chemenu.config`:
`chemenu.telemetry.policy` reads its record, and a hook handler imports that on
every tool call without the venv.
"""
from __future__ import annotations

import json
from dataclasses import dataclass, field as dc_field
from typing import Any, Callable, Optional

from chemenu import config

# The three tiers a settings field can belong to (`chemenu.settings`).
INSTANCE = "instance"
CHECKOUT = "checkout"
SECRET = "secret"
TIERS = (INSTANCE, CHECKOUT, SECRET)

_ABSENT = object()


# ---------------------------------------------------------------------------
# The record types
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class Field:
    """One key of a configuration object.

    `json_type` is the JSON type a conforming file uses (`boolean`, `integer`,
    `string`, `array`, `object`, or None for any). `coerce` is what the loader
    applies to the value it finds - it may accept more than `json_type`, which
    is the leniency `deviations()` reports - and raises `TypeError`/`ValueError`
    on a value it cannot use. `check` then judges the coerced value and returns
    a problem, or None.

    The last four attributes are the field's place in the tiered settings
    (`chemenu.settings`). `tier` says which file the field's value lives in.
    `private` marks private infrastructure - a host name, an internal URL - that
    may sit in the committed `wikitool.json` only as a reference, never as
    plain text. `names_command` and `gate` mark a field that names a program to
    run or that loosens or arms a gate. Such a field is never `instance`: a
    `git pull` from another clone must not change what `wikitool` executes or
    where it pushes. That rule is enforced here, when the field is declared,
    rather than by whoever reads it."""

    name: str
    type: str
    description: str
    required: bool = False
    default: Any = None
    default_text: str = ""
    json_type: Optional[str] = None
    coerce: Optional[Callable[[Any], Any]] = None
    check: Optional[Callable[[Any], Optional[str]]] = None
    choices: tuple[str, ...] = ()
    empty_is_missing: bool = False
    required_why: str = ""
    lenient: str = ""
    tier: str = ""
    private: bool = False
    names_command: bool = False
    gate: bool = False

    def __post_init__(self) -> None:
        where = f"config_shapes: field {self.name!r}"
        if self.tier and self.tier not in TIERS:
            raise ValueError(f"{where}: tier must be one of {', '.join(TIERS)}, not {self.tier!r}")
        if not self.tier and (self.private or self.names_command or self.gate):
            raise ValueError(f"{where}: `private`, `names_command` and `gate` need a tier")
        if self.tier == INSTANCE and self.user_only:
            raise ValueError(
                f"{where}: a field that names a program to run or affects a gate is never "
                f"tier {INSTANCE!r} - a pull from another clone would change what runs or "
                f"where it pushes; declare it {CHECKOUT!r}"
            )
        if self.private and self.tier != CHECKOUT:
            raise ValueError(
                f"{where}: a private field is tier {CHECKOUT!r} - its plain value never goes "
                "into the committed file, only a reference to it may"
            )

    @property
    def user_only(self) -> bool:
        """Set only by the user's own edit, never through `wikitool config set`."""
        return self.names_command or self.gate

    @property
    def takes_ref(self) -> bool:
        """Whether a `{"ref": ...}` may stand in for the value - required for a
        secret, optional for a private field, refused everywhere else."""
        return self.tier == SECRET or self.private


@dataclass(frozen=True)
class Shape:
    """One settings section: `key` is its key in `wikitool.json` and
    `.wikitool-local.json`, `fields` are its fields, each with its tier.
    `absent` and `malformed` say what a missing section and a broken one mean
    to the feature that reads it."""

    key: str
    title: str
    purpose: str
    fields: tuple[Field, ...]
    examples: tuple[dict, ...]
    absent: str = ""
    malformed: str = ""

    def field(self, name: str) -> Field:
        for candidate in self.fields:
            if candidate.name == name:
                return candidate
        raise KeyError(name)


@dataclass(frozen=True)
class Problem:
    field: str
    message: str


@dataclass
class Checked:
    """What `check()` found: the coerced values (defaults filled in), the
    fields the file actually set and the loader accepted, the problems a
    loader that rejects would reject on, the fields accepted only by coercion,
    and the keys no field names."""

    values: dict = dc_field(default_factory=dict)
    given: set = dc_field(default_factory=set)
    problems: list = dc_field(default_factory=list)
    coerced: list = dc_field(default_factory=list)
    unknown: list = dc_field(default_factory=list)

    def message(self) -> str:
        return "; ".join(p.message for p in self.problems)


# ---------------------------------------------------------------------------
# Coercions and conditions
# ---------------------------------------------------------------------------


def exact_str(value: Any) -> str:
    if not isinstance(value, str):
        raise TypeError("not a string")
    return value


def str_list(value: Any) -> list[str]:
    if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
        raise TypeError("not a list of strings")
    return list(value)


def extensions(value: Any) -> tuple[str, ...]:
    if not isinstance(value, list):
        raise TypeError("not a list")
    return tuple(sorted({str(ext).lower() for ext in value}))


def _positive(value: int) -> Optional[str]:
    return None if value > 0 else "must be positive"


def dotted_extensions(value: tuple[str, ...]) -> Optional[str]:
    if not value:
        return "must not be empty"
    not_dotted = [ext for ext in value if not ext.startswith(".")]
    return f"must each start with '.': {not_dotted}" if not_dotted else None


def positive_integer(name: str, description: str, **kwargs: Any) -> Field:
    return Field(
        name, "positive integer", description, json_type="integer", coerce=int,
        check=_positive, lenient="A number written as a string is accepted.", **kwargs,
    )


def _required_string(name: str, description: str, **kwargs: Any) -> Field:
    return Field(name, "string", description, required=True, json_type="string", coerce=exact_str,
                 empty_is_missing=True, **kwargs)


# ---------------------------------------------------------------------------
# The sections
# ---------------------------------------------------------------------------

PROVIDERS = ("superproductivity", "caldav")
SUPERPRODUCTIVITY_ACCESS = ("api", "snapshot")

TELEMETRY = Shape(
    key="telemetry",
    title="Telemetry",
    purpose="Turns telemetry on or off for this checkout, against the installation-form default, "
    "and sets its two quantity caps.",
    fields=(
        Field(
            "enabled", "boolean", "Telemetry on or off. `WIKI_TRACE` beats it in both directions.",
            default_text="the installation-form default", json_type="boolean", coerce=bool,
            lenient="Any other value is read by its truthiness, so `\"no\"` means on.",
            tier=CHECKOUT,
        ),
        positive_integer(
            "max_session_bytes", "Byte cap per session trace. `WIKI_TRACE_MAX_SESSION_BYTES` beats it.",
            default=5 * 1024 * 1024, tier=CHECKOUT,
        ),
        positive_integer(
            "keep_sessions", "How many session directories retention keeps. "
            "`WIKI_TRACE_KEEP_SESSIONS` beats it.", default=250, tier=CHECKOUT,
        ),
    ),
    examples=({"enabled": True, "max_session_bytes": 5242880, "keep_sessions": 250},),
    absent="The installation-form default applies: on in a clone of the origin repository, off in "
    "an instance.",
    malformed="The hook that reads it never fails on it: a value it cannot use counts as absent, "
    "and an unreadable `.wikitool-local.json` as an absent section. `doctor` reports each such "
    "deviation as a `WARN` under `telemetry`, naming the field.",
)

REMOTES = Shape(
    key="remotes",
    title="Allowed push URLs",
    purpose="Restricts the URLs `publish` may push to.",
    fields=(
        Field(
            "allowed_push_urls", "array of strings", "The push URLs `publish` may write to, "
            "compared with the URL the remote resolves to for a push - not with its name. An "
            "empty list allows no push at all.",
            required=True, json_type="array", coerce=str_list, tier=CHECKOUT, gate=True,
        ),
    ),
    examples=({"allowed_push_urls": ["<url>"]},),
    absent="Any remote is allowed; the Publish-Remote Gate does not apply.",
    malformed="`publish` stops with an error before committing anything. A broken section is never "
    "read as \"no restriction\".",
)

UPLOAD = Shape(
    key="upload",
    title="MCP submissions",
    purpose="Turns on the MCP read server's `submit` tool and sets its limits.",
    fields=(
        Field(
            "identity_header", "string", "The request header the authenticating middleware sets to "
            "the submitter's identity on every request.",
            default="X-Forwarded-User", json_type="string", coerce=str, empty_is_missing=True,
            lenient="An empty value means the default.", tier=CHECKOUT, gate=True,
        ),
        positive_integer("max_bytes", "Largest single submission, in bytes.", required=True,
                         tier=CHECKOUT, gate=True),
        Field(
            "allowed_extensions", "array of strings", "The file extensions `submit` accepts, each "
            "starting with `.`. Compared case-insensitively.",
            required=True, json_type="array", coerce=extensions, check=dotted_extensions,
            lenient="Entries are read as strings and lower-cased; duplicates collapse.",
            tier=CHECKOUT, gate=True,
        ),
        positive_integer("quota_submissions_per_day", "Submissions one submitter may make over any "
                         "rolling 24 hours.", required=True, tier=CHECKOUT, gate=True),
        positive_integer("quota_bytes_per_day", "Bytes one submitter may submit over any rolling "
                         "24 hours.", required=True, tier=CHECKOUT, gate=True),
    ),
    examples=({
        "identity_header": "X-Forwarded-User", "max_bytes": 10485760,
        "allowed_extensions": [".md", ".pdf"],
        "quota_submissions_per_day": 20, "quota_bytes_per_day": 52428800,
    },),
    absent="The `submit` tool is not registered: the write path does not exist.",
    malformed="An error. The MCP server does not start and `doctor` reports `FAIL`. A broken section "
    "is never read as \"no limits\".",
)

TASKS = Shape(
    key="tasks",
    title="Task tracker",
    purpose="Picks the task tracker the weekly review reads and writes through and sets the "
    "review's thresholds. The provider's own connection settings sit in a section of the same "
    "name.",
    fields=(
        Field("provider", "string", "Which adapter the review goes through.",
              required=True, json_type="string", coerce=str, choices=PROVIDERS, tier=INSTANCE),
        positive_integer("stalled_waiting_days", "A waiting-for item older than this many days "
                         "counts as overdue.", required=True, tier=INSTANCE),
        positive_integer("unpaged_project_weeks", "A tracker project with no `kb/` page is "
                         "reported once it is older than this many weeks.", required=True,
                         tier=INSTANCE),
        positive_integer("someday_stale_months", "A someday/maybe item untouched for this many "
                         "months counts as stale.", required=True, tier=INSTANCE),
    ),
    examples=({
        "provider": "superproductivity", "stalled_waiting_days": 14, "unpaged_project_weeks": 3,
        "someday_stale_months": 5,
    },),
    absent="No tracker is configured - a valid state. The weekly review has nothing to check "
    "against.",
    malformed="An error, and `doctor` reports `FAIL`. A broken section is never read as \"no tracker "
    "configured\".",
)

SUPERPRODUCTIVITY = Shape(
    key="superproductivity",
    title="Super Productivity",
    purpose="Connection settings for the Super Productivity adapter. `access` decides which of "
    "the other fields apply: `api_base_url` and `api_token` with `api`, `backups_dir` with "
    "`snapshot`.",
    fields=(
        Field(
            "access", "string", "Which way in: `api`, the local REST API (reads the current state "
            "and carries the write path), or `snapshot`, the newest backup snapshot (read-only).",
            required=True, json_type="string", choices=SUPERPRODUCTIVITY_ACCESS, tier=INSTANCE,
            required_why="there is no default and no fallback between the two",
        ),
        Field(
            "api_base_url", "string", "With `access: \"api\"`: where the app's REST API listens.",
            default="http://127.0.0.1:3876", json_type="string", coerce=str,
            empty_is_missing=True, lenient="An empty value means the default.",
            tier=CHECKOUT, private=True,
        ),
        Field(
            "api_token", "string", "With `access: \"api\"`, required: the token from the app's "
            "settings. Only ever a reference.", json_type="string", coerce=exact_str,
            empty_is_missing=True, tier=SECRET,
            required_why="every endpoint but GET /health requires Authorization: Bearer <token>",
        ),
        Field(
            "backups_dir", "string", "With `access: \"snapshot\"`, required: the directory the app "
            "writes its backup snapshots to; `~` is expanded.",
            json_type="string", coerce=exact_str, empty_is_missing=True, tier=CHECKOUT,
            required_why="the read path has nothing to read otherwise",
        ),
    ),
    examples=(
        {"access": "api", "api_base_url": "http://127.0.0.1:3876", "api_token": {"ref": "sp_token"}},
        {"access": "snapshot", "backups_dir": "~/.config/superProductivity/backups"},
    ),
    absent="Read only when `tasks.provider` is `superproductivity`.",
    malformed="An error when it is read: a field of the other access path beside `access`, or a "
    "missing `api_token` or `backups_dir`, stops the command with exit 1.",
)

CALDAV = Shape(
    key="caldav",
    title="CalDAV",
    purpose="Connection settings for the CalDAV adapter.",
    fields=(
        _required_string("url", "The account's CalDAV calendar home set. A trailing `/` is added "
                         "when missing.", tier=CHECKOUT, private=True),
        _required_string("username", "The login name, which can differ from the user id in the "
                         "URL.", tier=CHECKOUT, private=True),
        Field("app_password", "string", "An app password is recommended over the account "
              "password. Only ever a reference.", required=True, json_type="string", coerce=exact_str,
              empty_is_missing=True, tier=SECRET),
        _required_string("inbox_list", "The name of the one inbox list.", tier=INSTANCE),
        _required_string("someday_list", "The name of the one someday/maybe list.", tier=INSTANCE),
        Field("exclude_lists", "array of strings", "Existing plain task lists that are not "
              "projects.", default=[], json_type="array", coerce=str_list, tier=INSTANCE),
    ),
    examples=({
        "url": {"ref": "op://Chemenu/caldav/url"}, "username": "<login>",
        "app_password": {"ref": "caldav_app_password"}, "inbox_list": "Inbox",
        "someday_list": "Someday", "exclude_lists": ["..."],
    },),
    absent="Read only when `tasks.provider` is `caldav`.",
)

# The sections of the tiered settings files, `wikitool.json` and
# `.wikitool-local.json` (`chemenu.settings`), read at call time so a test can
# stand in a section of its own.
SETTINGS_SECTIONS: tuple[Shape, ...] = (TELEMETRY, REMOTES, UPLOAD, TASKS, SUPERPRODUCTIVITY, CALDAV)

# The sections `WIKITOOL_TASKS_CONFIG` replaces as a whole: a profile names a
# tracker, and everything that configures one.
PROFILE_SECTIONS = ("tasks", "superproductivity", "caldav")


# ---------------------------------------------------------------------------
# Checking a value against a record
# ---------------------------------------------------------------------------

_JSON_TYPES: dict[str, Callable[[Any], bool]] = {
    "boolean": lambda v: isinstance(v, bool),
    "integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
    "string": lambda v: isinstance(v, str),
    "array": lambda v: isinstance(v, list),
    "object": lambda v: isinstance(v, dict),
}

_TYPE_NAMES = {bool: "a boolean", int: "an integer", float: "a number", str: "a string",
               list: "an array", dict: "an object", type(None): "null"}


def _described(value: Any) -> str:
    return _TYPE_NAMES.get(type(value), type(value).__name__)


def _article(noun: str) -> str:
    return f"{'an' if noun[:1] in 'aeiou' else 'a'} {noun}"


def check(shape: Shape, data: Any) -> Checked:
    """`data` read against `shape`: coerced values, problems, and what was
    only accepted leniently. Never raises - each loader decides what a problem
    means for it. A key no field names is listed in `unknown`, not a problem."""
    out = Checked()
    if not isinstance(data, dict):
        out.problems.append(Problem("", f"must contain a JSON object, not {_described(data)}"))
        return out
    for spec in shape.fields:
        raw = data.get(spec.name, _ABSENT)
        if raw is _ABSENT or (spec.empty_is_missing and not raw):
            if spec.required:
                why = f" - {spec.required_why}" if spec.required_why else ""
                empty = " or empty" if spec.empty_is_missing else ""
                out.problems.append(Problem(spec.name, f"{spec.name} is missing{empty}, and required{why}"))
            else:
                out.values[spec.name] = spec.default
            continue
        try:
            value = spec.coerce(raw) if spec.coerce is not None else raw
        except (TypeError, ValueError):
            out.problems.append(Problem(
                spec.name, f"{spec.name} must be {_article(spec.type)}, got {_described(raw)}"))
            continue
        if spec.choices and value not in spec.choices:
            why = f" - {spec.required_why}" if spec.required_why else ""
            out.problems.append(Problem(
                spec.name,
                f"{spec.name} must be one of {', '.join(repr(c) for c in spec.choices)}{why}; got {raw!r}",
            ))
            continue
        problem = spec.check(value) if spec.check is not None else None
        if problem:
            out.problems.append(Problem(spec.name, f"{spec.name} {problem}"))
            continue
        if spec.json_type is not None and not _JSON_TYPES[spec.json_type](raw):
            out.coerced.append(Problem(
                spec.name, f"{spec.name} is {_described(raw)}, not a JSON {spec.json_type}, and is "
                f"read as {value!r}",
            ))
        out.values[spec.name] = value
        out.given.add(spec.name)
    out.unknown = sorted(set(data) - {spec.name for spec in shape.fields})
    return out


def deviations(shape: Shape, data: Any) -> list[str]:
    """Everything in `data` that a strict reading of `shape` would not take as
    written: problems, values accepted only by coercion, and unknown keys. For
    a section whose loader is silent about all three - telemetry - this is what
    `doctor` reports."""
    checked = check(shape, data)
    found = [p.message for p in checked.problems] + [p.message for p in checked.coerced]
    found += [f"unknown key {key!r} is ignored" for key in checked.unknown]
    return found


# ---------------------------------------------------------------------------
# Rendering
# ---------------------------------------------------------------------------


def _cell(text: str) -> str:
    return text.replace("|", "\\|").replace("\n", " ")


def _default_cell(spec: Field) -> str:
    if spec.required:
        return "-"
    if spec.default_text:
        return spec.default_text
    if spec.default is None:
        return "-"
    return f"`{json.dumps(spec.default)}`"


def _tier_cell(spec: Field) -> str:
    flags = [name for name, on in (("private", spec.private), ("names a command", spec.names_command),
                                   ("gate", spec.gate)) if on]
    return ", ".join([spec.tier, *flags])


def render_markdown(shape: Shape) -> str:
    """The reference text for one section: its fields with their tiers, what
    an absent and a malformed section mean, and an example."""
    lines = [
        "| Field | Tier | Type | Required | Default | Description |",
        "|-------|------|------|----------|---------|-------------|",
    ]
    for spec in shape.fields:
        required = "yes" if spec.required else "no"
        if spec.required and spec.empty_is_missing:
            required = "yes, not empty"
        description = " ".join(part for part in (spec.description, spec.lenient) if part)
        lines.append(
            f"| `{spec.name}` | {_tier_cell(spec)} | {spec.type} | {required} | "
            f"{_cell(_default_cell(spec))} | {_cell(description)} |"
        )
    lines.append("")
    if shape.absent:
        lines.append(f"- **Absent:** {shape.absent}")
    if shape.malformed:
        lines.append(f"- **Malformed:** {shape.malformed}")
    lines += ["", "Example:" if len(shape.examples) == 1 else "Examples:", ""]
    for example in shape.examples:
        lines += ["```json", json.dumps({shape.key: example}, indent=2), "```", ""]
    return "\n".join(lines).rstrip("\n") + "\n"