From 8d4a2eb6528d7e5acd8ebef9dc12550dd3df8ac8 Mon Sep 17 00:00:00 2001 From: dlxeva Date: Fri, 11 Sep 2026 03:52:17 +0800 Subject: [PATCH] feat: add source-backed work view --- docs/product/context-pack-contract.md | 46 +++ src/flg/commands/context.py | 89 +++++- src/flg/commands/doctor.py | 16 +- src/flg/core/work_view.py | 425 ++++++++++++++++++++++++++ tests/test_work_view.py | 163 ++++++++++ 5 files changed, 733 insertions(+), 6 deletions(-) create mode 100644 src/flg/core/work_view.py create mode 100644 tests/test_work_view.py diff --git a/docs/product/context-pack-contract.md b/docs/product/context-pack-contract.md index df33383..b9b2f11 100644 --- a/docs/product/context-pack-contract.md +++ b/docs/product/context-pack-contract.md @@ -28,6 +28,52 @@ For a smaller navigation-first entrypoint, use: flg context --mode manifest ``` +For projects whose live work queue is maintained in one authoritative source +block, declare a `work_source` extension in `.flg/state.json` and build the +derived view: + +```json +{ + "work_source": { + "schema_version": "1", + "path": "docs/work-ledger.md", + "start_marker": "", + "end_marker": "" + } +} +``` + +```bash +flg context --mode work +flg doctor --strict +``` + +The generated `.flg/context/work-view.md` is not another writable ledger. It +is a bounded projection containing the current action, blockers, necessary +constraints, source locator, and marked-block SHA. Changes outside the marked +block do not invalidate it. Changes inside the block make the prior view stale +and remove its action from canonical continuation until a human or agent +rechecks the source and rebuilds the view. Omitting both markers intentionally +uses the whole file. Source paths must remain inside the project and symlinks +are rejected. + +The generic extractor recognizes these headings inside the declared block: + +```markdown +## Current Action +- Continue the one action that is safe to inherit. + +## Blockers +- Record a blocker, or an explicit `none` item. + +## Necessary Constraints +- Record only constraints needed for the current action. +``` + +Project-specific task tables may keep their own adapter and still use the same +block-SHA freshness contract. FlowGrid core does not infer proprietary IDs, +owners, or table semantics from arbitrary columns. + The generated Continuity Manifest contains identity, the current goal, judgment statuses and IDs, active-work pointers, source health, and exact commands for expanding a judgment through the existing `evidence` and `trace` paths. It is a diff --git a/src/flg/commands/context.py b/src/flg/commands/context.py index 67a5f1e..184057f 100644 --- a/src/flg/commands/context.py +++ b/src/flg/commands/context.py @@ -17,6 +17,7 @@ from ..core.files import is_flg_project, read_file_safe from ..core.state import load_state from ..core.wiki import wiki_context_summary +from ..core.work_view import build_work_view, inspect_work_view, write_work_view_manifest from .handoff import parse_patch_for_handoff console = Console() @@ -428,6 +429,52 @@ def _render_source_health(report: dict) -> str: return "\n".join(lines) + "\n" +def _work_source_current_action(current_action: dict, work_view: dict | None) -> dict: + """Use an explicitly declared, SHA-checked work source when configured.""" + if work_view is None: + return current_action + return { + "status": work_view["action_status"], + "action": ( + work_view.get("current_action") + if work_view["action_status"] == "current" + else None + ), + "source": f"{work_view['source_path']} ({work_view['locator']})", + "source_updated_at": work_view.get("generated_at", "not generated"), + "ignored_fallback_count": current_action.get("ignored_fallback_count", 0), + "reason": work_view["reason"], + } + + +def _render_work_view(work_view: dict | None) -> str: + if work_view is None: + return "- Status: not configured\n" + blockers = work_view.get("blockers") or [] + constraints = work_view.get("constraints") or [] + missing = work_view.get("missing") or [] + lines = [ + f"- Status: {work_view['status']}", + f"- Action status: {work_view['action_status']}", + f"- Source path: {work_view['source_path']}", + f"- Locator: {work_view['locator']}", + f"- Current block SHA-256: {work_view['current_block_sha256']}", + ] + if work_view.get("recorded_block_sha256"): + lines.append(f"- Recorded block SHA-256: {work_view['recorded_block_sha256']}") + lines.extend( + ( + f"- Blockers: {'; '.join(blockers) if blockers else '(not recorded)'}", + f"- Necessary constraints: {'; '.join(constraints) if constraints else '(not recorded)'}", + f"- Missing information: {'; '.join(missing) if missing else '(none detected)'}", + f"- Reason: {work_view['reason']}", + "- Inspect source: open the path at the locator above before refreshing.", + "- Refresh after recheck: `flg context --mode work`", + ) + ) + return "\n".join(lines) + "\n" + + def _render_current_action(current_action: dict) -> str: """Render the same canonical action contract in every continuation view.""" action = current_action.get("action") or "(none; reconcile state before acting)" @@ -539,12 +586,14 @@ def _build_continuity_manifest(root: Path, budget: int) -> tuple[str, dict]: pending_patches = _pending_patch_summaries(root) pending_captures = _pending_capture_ids(root) source_health = validate_project(root) + work_view = inspect_work_view(root, state) current_action = resolve_current_action( snapshot_content, state, pending_patches_count=len(pending_patches), framing_goal_defined=framing_goal_defined, ) + current_action = _work_source_current_action(current_action, work_view) pending_decision_ids = [ decision["decision_id"] @@ -606,6 +655,11 @@ def _build_continuity_manifest(root: Path, budget: int) -> tuple[str, dict]: ## Current Action {_render_current_action(current_action)}""" + work_view_section = f""" + +## Source-backed Work View + +{_render_work_view(work_view)}""" work_pointer_section = f""" ## Work Pointers @@ -636,6 +690,7 @@ def _build_continuity_manifest(root: Path, budget: int) -> tuple[str, dict]: content = ( identity_section + current_action_section + + work_view_section + judgment_section + work_pointer_section + source_health_section @@ -655,6 +710,7 @@ def _build_continuity_manifest(root: Path, budget: int) -> tuple[str, dict]: content = ( identity_section + current_action_section + + work_view_section + compact_judgment_section + work_pointer_section + source_health_section @@ -686,16 +742,31 @@ def _build_continuity_manifest(root: Path, budget: int) -> tuple[str, dict]: ), "source_health": source_health, "current_action": current_action, + "work_view": work_view, "truncated": truncated, } return content, metadata def build_context_pack(root: Path, mode: str = "resume", budget: int = 4000) -> tuple[str, dict]: + if mode == "work": + state = load_state(root) + if not state: + raise ValueError("No readable state found. Run 'flg init' first.") + content, metadata = build_work_view(root, state, budget=budget) + metadata.update( + { + "path": str(root / ".flg" / "context" / "work-view.md"), + "sources_included": [".flg/state.json#work_source", metadata["source"]["path"]], + "pending_patches_count": 0, + "confirmed_decisions_count": 0, + } + ) + return content, metadata if mode == "manifest": return _build_continuity_manifest(root, budget) if mode != "resume": - raise ValueError("Supported context modes: resume, manifest") + raise ValueError("Supported context modes: resume, manifest, work") state = load_state(root) if not state: @@ -746,12 +817,14 @@ def build_context_pack(root: Path, mode: str = "resume", budget: int = 4000) -> confirmed_decisions = _parse_confirmed_decisions(decisions_content, evidence_items) pending_patches = _pending_patch_summaries(root) source_health = validate_project(root) + work_view = inspect_work_view(root, state) current_action = resolve_current_action( snapshot_content, state, pending_patches_count=len(pending_patches), framing_goal_defined=bool(framing_goal), ) + current_action = _work_source_current_action(current_action, work_view) assumptions = _list_items(_section(snapshot_content, "Unconfirmed"), limit=8) assumptions += _list_items(_section(snapshot_content, "未确认"), limit=8) @@ -869,6 +942,9 @@ def build_context_pack(root: Path, mode: str = "resume", budget: int = 4000) -> ## Current Action {_render_current_action(current_action)} +## Source-backed Work View + +{_render_work_view(work_view)} ## Project Frame {project_frame} @@ -950,6 +1026,7 @@ def build_context_pack(root: Path, mode: str = "resume", budget: int = 4000) -> "confirmed_decisions_count": len(confirmed_decisions), "source_health": source_health, "current_action": current_action, + "work_view": work_view, "wiki": wiki_summary, "wiki_truncated": wiki_truncated, "truncated": truncated, @@ -958,7 +1035,7 @@ def build_context_pack(root: Path, mode: str = "resume", budget: int = 4000) -> def context_command( - mode: str = typer.Option("resume", "--mode", help="Context mode: 'resume' or compact 'manifest'."), + mode: str = typer.Option("resume", "--mode", help="Context mode: 'resume', compact 'manifest', or source-backed 'work'."), budget: int = typer.Option(4000, "--budget", help="Approximate token budget for the generated context pack."), output: Optional[str] = typer.Option(None, "--output", "-o", help="Optional output path. Defaults by mode under .flg/context/."), print_pack: bool = typer.Option(False, "--print", help="Print the generated context pack after writing it."), @@ -975,22 +1052,24 @@ def context_command( console.print(f"[red]{exc}[/red]") raise typer.Exit(1) from exc - default_name = "manifest.md" if mode == "manifest" else "startup.md" + default_name = {"manifest": "manifest.md", "work": "work-view.md"}.get(mode, "startup.md") output_path = Path(output) if output else root / ".flg" / "context" / default_name if not output_path.is_absolute(): output_path = root / output_path output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(content, encoding="utf-8") + if mode == "work": + write_work_view_manifest(root, metadata) console.print() - artifact_name = "Continuity Manifest" if mode == "manifest" else "Context Pack" + artifact_name = {"manifest": "Continuity Manifest", "work": "Source-backed Work View"}.get(mode, "Context Pack") console.print(f"[bold green]✓ {artifact_name} generated[/bold green]") console.print(f"[bold]Path:[/bold] {output_path}") console.print(f"[bold]Size:[/bold] {metadata['chars']} chars (~{metadata['estimated_tokens']} tokens)") console.print(f"[bold]Sources included:[/bold] {len(metadata['sources_included'])}") console.print(f"[bold]Pending patches:[/bold] {metadata['pending_patches_count']}") console.print(f"[bold]Confirmed decisions:[/bold] {metadata['confirmed_decisions_count']}") - if metadata["confirmed_decisions_count"] == 0: + if mode != "work" and metadata["confirmed_decisions_count"] == 0: console.print(f"[yellow]Warning: no reviewed decisions found. {artifact_name} will rely on current state and pending material.[/yellow]") if metadata["truncated"]: if mode == "manifest": diff --git a/src/flg/commands/doctor.py b/src/flg/commands/doctor.py index 309e786..823c1c7 100644 --- a/src/flg/commands/doctor.py +++ b/src/flg/commands/doctor.py @@ -14,6 +14,7 @@ from ..core.relations import validate_decision_relations from ..core.state import load_state from ..core.wiki import wiki_health_issues +from ..core.work_view import work_view_health_issues console = Console() @@ -102,8 +103,10 @@ def doctor( relation_issues = validate_decision_relations(decisions_content) identity = _runtime_identity(root) identity_issues = identity.get("issues", []) if identity else [] - delivery_issues = active_delivery_issues(load_state(root) or {}) + state = load_state(root) or {} + delivery_issues = active_delivery_issues(state) wiki_issues = wiki_health_issues(root) + work_view_issues = work_view_health_issues(root, state) table = Table(title=f"FlowGrid Doctor: {root}") table.add_column("Check", style="cyan") table.add_column("Result", style="bold") @@ -113,6 +116,7 @@ def doctor( and not identity_issues and not delivery_issues and not wiki_issues + and not work_view_issues ) table.add_row("Overall", "OK" if overall_ok else "Needs attention") table.add_row("Formal decisions", str(report["decision_count"])) @@ -159,6 +163,11 @@ def doctor( "not configured" if not (root / ".flg" / "wiki.json").exists() else ("OK" if not wiki_issues else f"Needs attention ({len(wiki_issues)})"), ) + table.add_row( + "Source-backed work view", + "not configured" if "work_source" not in state + else ("OK" if not work_view_issues else f"Needs attention ({len(work_view_issues)})"), + ) if identity is None: table.add_row("Runtime identity", "not configured (no repo-map)") elif identity.get("error"): @@ -219,6 +228,11 @@ def doctor( for issue in wiki_issues: console.print(f" - {issue}") + if work_view_issues: + console.print("[yellow]source_backed_work_view:[/yellow]") + for issue in work_view_issues: + console.print(f" - {issue}") + if strict and not overall_ok: raise typer.Exit(1) diff --git a/src/flg/core/work_view.py b/src/flg/core/work_view.py new file mode 100644 index 0000000..34fe9b7 --- /dev/null +++ b/src/flg/core/work_view.py @@ -0,0 +1,425 @@ +"""Build and inspect a bounded view over a declared current-work source.""" + +from __future__ import annotations + +import hashlib +import json +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +WORK_VIEW_MANIFEST = ".flg/context/work-view.json" +WORK_VIEW_SCHEMA_VERSION = "1" +_MAX_SOURCE_CHARS = 1_000_000 + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _one_line(value: Any, limit: int = 240) -> str: + compact = re.sub(r"\s+", " ", str(value or "")).strip() + return compact.replace("`", "'")[:limit] + + +def _resolve_source(root: Path, value: Any) -> tuple[Path, str]: + if not isinstance(value, str) or not value.strip(): + raise ValueError("work_source.path must be a non-empty project-relative path.") + relative = Path(value) + if relative.is_absolute() or ".." in relative.parts: + raise ValueError(f"work_source.path must stay inside the project: {value}") + + root = root.resolve() + current = root + for part in relative.parts: + current = current / part + if current.is_symlink(): + raise ValueError(f"Symlinked work sources are not supported: {value}") + try: + resolved = (root / relative).resolve(strict=True) + except FileNotFoundError as exc: + raise ValueError(f"Declared work source does not exist: {value}") from exc + if root not in resolved.parents: + raise ValueError(f"work_source.path must stay inside the project: {value}") + if not resolved.is_file(): + raise ValueError(f"Declared work source must be a file: {value}") + return resolved, resolved.relative_to(root).as_posix() + + +def _marker(config: dict[str, Any], key: str) -> str | None: + value = config.get(key) + if value is None: + return None + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"work_source.{key} must be a non-empty string when provided.") + if "\n" in value or "\r" in value: + raise ValueError(f"work_source.{key} must be a single-line marker.") + if len(value) > 200: + raise ValueError(f"work_source.{key} must be at most 200 characters.") + return value + + +def load_work_source(root: Path, state: dict[str, Any] | None) -> dict[str, Any] | None: + """Validate the optional declaration stored as a state extension.""" + if not state or "work_source" not in state: + return None + raw = state["work_source"] + if not isinstance(raw, dict): + raise ValueError("work_source must be an object in .flg/state.json.") + schema = str(raw.get("schema_version", WORK_VIEW_SCHEMA_VERSION)) + if schema != WORK_VIEW_SCHEMA_VERSION: + raise ValueError(f"Unsupported work_source schema version: {schema}") + source, relative = _resolve_source(root, raw.get("path")) + start = _marker(raw, "start_marker") + end = _marker(raw, "end_marker") + if (start is None) != (end is None): + raise ValueError("work_source.start_marker and end_marker must be provided together.") + if start is not None and start == end: + raise ValueError("work_source start_marker and end_marker must be different.") + return { + "schema_version": schema, + "path": relative, + "resolved_path": source, + "start_marker": start, + "end_marker": end, + } + + +def _source_block(config: dict[str, Any]) -> dict[str, Any]: + path: Path = config["resolved_path"] + try: + text = path.read_text(encoding="utf-8") + except UnicodeError as exc: + raise ValueError( + f"Declared work source must be UTF-8 text: {config['path']}" + ) from exc + if len(text) > _MAX_SOURCE_CHARS: + raise ValueError( + f"Declared work source exceeds {_MAX_SOURCE_CHARS} characters: {config['path']}" + ) + lines = text.splitlines(keepends=True) + start = config["start_marker"] + end = config["end_marker"] + if start is None: + block = text + start_line = 1 + end_line = max(1, len(text.splitlines())) + locator = config["path"] + else: + start_hits = [idx for idx, line in enumerate(lines) if line.rstrip("\r\n") == start] + end_hits = [idx for idx, line in enumerate(lines) if line.rstrip("\r\n") == end] + if len(start_hits) != 1: + raise ValueError( + f"start_marker must occur exactly once in {config['path']}; " + f"found {len(start_hits)}." + ) + if len(end_hits) != 1: + raise ValueError( + f"end_marker must occur exactly once in {config['path']}; found {len(end_hits)}." + ) + if start_hits[0] >= end_hits[0]: + raise ValueError(f"work_source markers are out of order in {config['path']}.") + block = "".join(lines[start_hits[0] + 1 : end_hits[0]]) + start_line = start_hits[0] + 2 + end_line = max(start_line, end_hits[0]) + locator = f"{config['path']}#L{start_line}-L{end_line}" + return { + "text": block, + "sha256": hashlib.sha256(block.encode("utf-8")).hexdigest(), + "locator": locator, + "start_line": start_line, + "end_line": end_line, + } + + +def _section(text: str, aliases: tuple[str, ...]) -> str: + lines = text.splitlines() + aliases_lower = {alias.casefold() for alias in aliases} + start: int | None = None + level = 0 + for idx, line in enumerate(lines): + match = re.match(r"^(#{1,6})\s+(.+?)\s*$", line) + if match and match.group(2).casefold() in aliases_lower: + start = idx + 1 + level = len(match.group(1)) + break + if start is None: + return "" + end = len(lines) + for idx in range(start, len(lines)): + match = re.match(r"^(#{1,6})\s+", lines[idx]) + if match and len(match.group(1)) <= level: + end = idx + break + return "\n".join(lines[start:end]).strip() + + +def _items(text: str, *, limit: int, item_chars: int) -> list[str]: + items: list[str] = [] + for raw in text.splitlines(): + line = re.sub(r"^\s*(?:[-*+]\s+|\d+[.)]\s+)", "", raw).strip() + line = re.sub(r"^#+\s+", "", line).strip() + if not line or line.startswith("\n" + if compacted + else "" + ) + return f"""# FLG Source-backed Work View + +## Freshness + +- Status: fresh +- Block SHA-256: {block['sha256']} +- Generated: {_now()} + +## Source + +- Path: {config['path']} +- Locator: {_one_line(block['locator'], 320)} +- Authority: declared current-work source; this view does not create or review formal decisions. + +## Current Action + +- Status: {action_status} +- Action: {fields['current_action'] or '(not defined)'} + +## Blockers + +{_render_list(fields['blockers'], '(not recorded)')} + +## Necessary Constraints + +{_render_list(fields['constraints'], '(not recorded)')} + +## Missing Information + +{_render_list(fields['missing'], '(none detected)')} + +## Expand On Demand + +- Inspect the declared source at the path and locator above before rebuilding. +- Full FlowGrid context: `flg context --mode resume --budget 4000` + +## Boundary + +- Rebuild with `flg context --mode work` after reviewing a changed source block. +- Candidate judgments still require the normal report-only review and review/merge gate. +{note}""" + + +def build_work_view( + root: Path, + state: dict[str, Any] | None, + budget: int = 1500, +) -> tuple[str, dict[str, Any]]: + config = load_work_source(root, state) + if config is None: + raise ValueError( + "No work source declared. Add work_source.path and optional " + "start_marker/end_marker to .flg/state.json." + ) + block = _source_block(config) + fields = _extract_fields(block["text"]) + max_chars = max(1200, min(max(1, budget) * 4, 6000)) + content = _render(config, block, fields, compacted=False) + truncated = False + if len(content) > max_chars: + fields = _extract_fields(block["text"], compact=True) + content = _render(config, block, fields, compacted=True) + truncated = True + if len(content) > max_chars: + raise ValueError( + f"Work View required fields exceed the {max_chars}-character budget; increase --budget." + ) + metadata = { + "schema_version": WORK_VIEW_SCHEMA_VERSION, + "generated_at": _now(), + "source": { + "path": config["path"], + "start_marker": config["start_marker"], + "end_marker": config["end_marker"], + "locator": block["locator"], + "block_sha256": block["sha256"], + "start_line": block["start_line"], + "end_line": block["end_line"], + }, + "current_action": fields["current_action"], + "blockers": fields["blockers"], + "constraints": fields["constraints"], + "missing": fields["missing"], + "status": "fresh", + "chars": len(content), + "estimated_tokens": len(content) // 4, + "truncated": truncated, + } + return content, metadata + + +def write_work_view_manifest(root: Path, metadata: dict[str, Any]) -> Path: + path = root / WORK_VIEW_MANIFEST + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(metadata, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + return path + + +def inspect_work_view(root: Path, state: dict[str, Any] | None) -> dict[str, Any] | None: + """Compare the generated view's block SHA with the current declared block.""" + config = load_work_source(root, state) + if config is None: + return None + block = _source_block(config) + path = root / WORK_VIEW_MANIFEST + base = { + "configured": True, + "source_path": config["path"], + "locator": block["locator"], + "current_block_sha256": block["sha256"], + } + if not path.exists(): + return { + **base, + "status": "unbuilt", + "action_status": "needs_recheck", + "current_action": None, + "blockers": [], + "constraints": [], + "missing": ["generated work view"], + "reason": "Run `flg context --mode work` to create the first SHA-backed view.", + } + try: + previous = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise ValueError(f"Work View manifest is corrupt: {path}") from exc + if previous.get("schema_version") != WORK_VIEW_SCHEMA_VERSION: + raise ValueError( + f"Unsupported Work View manifest schema: {previous.get('schema_version')}" + ) + old_source = previous.get("source") or {} + config_changed = any( + old_source.get(key) != config.get(key) + for key in ("path", "start_marker", "end_marker") + ) + stale = config_changed or old_source.get("block_sha256") != block["sha256"] + return { + **base, + "status": "stale" if stale else "fresh", + "action_status": ( + "needs_recheck" + if stale + else ("current" if previous.get("current_action") else "not_defined") + ), + "current_action": previous.get("current_action"), + "blockers": previous.get("blockers") or [], + "constraints": previous.get("constraints") or [], + "missing": previous.get("missing") or [], + "recorded_block_sha256": old_source.get("block_sha256"), + "generated_at": previous.get("generated_at", "unknown"), + "config_changed": config_changed, + "reason": ( + "Declared source configuration or marked block changed; rebuild only " + "after rechecking the source." + if stale + else "The declared source block matches the generated Work View SHA." + ), + } + + +def work_view_health_issues(root: Path, state: dict[str, Any] | None) -> list[str]: + """Return opt-in work-view issues for ``flg doctor``.""" + if not state or "work_source" not in state: + return [] + try: + status = inspect_work_view(root, state) + except (OSError, UnicodeError, ValueError) as exc: + return [f"source-backed work view invalid: {exc}"] + if status is None: + return [] + if status["status"] == "unbuilt": + return [ + "source-backed work view is unbuilt; review the source and run " + "`flg context --mode work`" + ] + if status["status"] == "stale": + return [ + "source-backed work view is stale; the declared source block changed " + "and requires recheck" + ] + return [] diff --git a/tests/test_work_view.py b/tests/test_work_view.py new file mode 100644 index 0000000..edd5afa --- /dev/null +++ b/tests/test_work_view.py @@ -0,0 +1,163 @@ +"""Source-backed current-work projection and block-level freshness tests.""" + +import json +import os + +import pytest +from typer.testing import CliRunner + +from flg.cli import app +from flg.commands.context import build_context_pack +from flg.core.state import load_state +from flg.core.work_view import build_work_view, inspect_work_view + + +runner = CliRunner() +START = "" +END = "" + + +def _project(tmp_path): + old = os.getcwd() + os.chdir(tmp_path) + try: + result = runner.invoke(app, ["init", "Generic Operations Project"]) + finally: + os.chdir(old) + assert result.exit_code == 0, result.output + source = tmp_path / "docs" / "work-ledger.md" + source.parent.mkdir(exist_ok=True) + source.write_text( + "# Work Ledger\n\nUnrelated preface.\n\n" + f"{START}\n" + "## Current Action\n- Interview the next operator.\n\n" + "## Blockers\n- Waiting for a sample export.\n\n" + "## Necessary Constraints\n- Do not treat a draft as approval.\n" + f"{END}\n\nUnrelated appendix.\n", + encoding="utf-8", + ) + state_path = tmp_path / ".flg" / "state.json" + state = json.loads(state_path.read_text(encoding="utf-8")) + state["work_source"] = { + "schema_version": "1", + "path": "docs/work-ledger.md", + "start_marker": START, + "end_marker": END, + } + state_path.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + return source + + +def _invoke(tmp_path, args): + old = os.getcwd() + os.chdir(tmp_path) + try: + return runner.invoke(app, args) + finally: + os.chdir(old) + + +def test_work_view_builds_generic_projection_and_tracks_only_marked_block(tmp_path): + source = _project(tmp_path) + result = _invoke(tmp_path, ["context", "--mode", "work"]) + assert result.exit_code == 0, result.output + rendered = (tmp_path / ".flg" / "context" / "work-view.md").read_text(encoding="utf-8") + assert "Interview the next operator." in rendered + assert "Waiting for a sample export." in rendered + assert "Do not treat a draft as approval." in rendered + assert "docs/work-ledger.md#L" in rendered + assert "sed -n" not in rendered + + state = load_state(tmp_path) + assert inspect_work_view(tmp_path, state)["status"] == "fresh" + + source.write_text(source.read_text(encoding="utf-8") + "Outside-only note.\n", encoding="utf-8") + assert inspect_work_view(tmp_path, state)["status"] == "fresh" + + source.write_text( + source.read_text(encoding="utf-8").replace( + "Interview the next operator.", "Interview two operators." + ), + encoding="utf-8", + ) + status = inspect_work_view(tmp_path, state) + assert status["status"] == "stale" + assert status["action_status"] == "needs_recheck" + _, manifest_meta = build_context_pack(tmp_path, mode="manifest") + assert manifest_meta["current_action"]["status"] == "needs_recheck" + assert manifest_meta["current_action"]["action"] is None + + +@pytest.mark.parametrize( + "replacement,error", + [ + ("", "start_marker must occur exactly once"), + (f"{START}\n{START}", "start_marker must occur exactly once"), + (f"{END}\n{START}", "markers are out of order"), + ], +) +def test_work_view_rejects_missing_duplicate_or_reversed_markers(tmp_path, replacement, error): + source = _project(tmp_path) + text = source.read_text(encoding="utf-8") + if replacement.startswith(END): + text = text.replace(START, "TEMP", 1).replace(END, START, 1).replace("TEMP", END, 1) + else: + text = text.replace(START, replacement, 1) + source.write_text(text, encoding="utf-8") + with pytest.raises(ValueError, match=error): + build_work_view(tmp_path, load_state(tmp_path)) + + +def test_work_view_rejects_path_escape_and_symlink(tmp_path): + _project(tmp_path) + state = load_state(tmp_path) + state["work_source"]["path"] = "../outside.md" + with pytest.raises(ValueError, match="stay inside"): + build_work_view(tmp_path, state) + + target = tmp_path / "docs" / "work-ledger.md" + link = tmp_path / "docs" / "linked.md" + link.symlink_to(target) + state["work_source"]["path"] = "docs/linked.md" + with pytest.raises(ValueError, match="Symlinked"): + build_work_view(tmp_path, state) + + +def test_work_view_rejects_non_utf8_source_with_actionable_error(tmp_path): + source = _project(tmp_path) + source.write_bytes(b"\xff\xfe\x00") + with pytest.raises(ValueError, match="must be UTF-8 text"): + build_work_view(tmp_path, load_state(tmp_path)) + + +def test_work_view_small_budget_compacts_once_without_recursive_failure(tmp_path): + source = _project(tmp_path) + text = source.read_text(encoding="utf-8") + long_items = "\n".join(f"- blocker-{index}-" + ("x" * 210) for index in range(6)) + text = text.replace("- Waiting for a sample export.", long_items) + source.write_text(text, encoding="utf-8") + content, metadata = build_work_view(tmp_path, load_state(tmp_path), budget=1) + assert len(content) <= 1200 + assert metadata["truncated"] is True + + +def test_doctor_strict_reports_unbuilt_and_stale_work_view(tmp_path): + source = _project(tmp_path) + unbuilt = _invoke(tmp_path, ["doctor", "--strict"]) + assert unbuilt.exit_code == 1 + assert "source-backed work view is unbuilt" in unbuilt.output + + built = _invoke(tmp_path, ["context", "--mode", "work"]) + assert built.exit_code == 0 + fresh = _invoke(tmp_path, ["doctor", "--strict"]) + assert "source-backed work view is stale" not in fresh.output + + source.write_text( + source.read_text(encoding="utf-8").replace( + "Waiting for a sample export.", "Waiting for two exports." + ), + encoding="utf-8", + ) + stale = _invoke(tmp_path, ["doctor", "--strict"]) + assert stale.exit_code == 1 + assert "source-backed work view is stale" in stale.output