diff --git a/planfile/core/configuration.py b/planfile/core/configuration.py index ddbec12..60afd3d 100644 --- a/planfile/core/configuration.py +++ b/planfile/core/configuration.py @@ -26,20 +26,12 @@ save_runtime_context_config, ) +from planfile.utils.yaml_cache import read_yaml_file as _read_yaml_file + if TYPE_CHECKING: from planfile import Planfile -@lru_cache(maxsize=256) -def _load_yaml_cached(path_str: str, mtime_ns: int) -> Any: - return yaml.safe_load(Path(path_str).read_text(encoding="utf-8")) - - -def _read_yaml_file(path: Path) -> Any: - stat = path.stat() - return _load_yaml_cached(str(path.resolve()), stat.st_mtime_ns) - - _SENSITIVE_PATH = re.compile( r"(^|[._-])(authorization|cookie|credential|password|passwd|secret|token|" r"api[_-]?key|private[_-]?key)($|[._-])", @@ -194,137 +186,53 @@ class Setting: description: str +_CFG_SRC = ".planfile/config.yaml" +_RT_SRC = ".koru/runtime-context.json" +_INT_SRC = f".planfile/{_INTEGRATION_CONFIG_NAME}" + STORE_SETTINGS: dict[str, Setting] = { - "store.project": Setting(_text, ".planfile/config.yaml", "Project display name"), - "store.prefix": Setting( - _prefix, - ".planfile/config.yaml", - "Prefix for newly allocated ticket IDs", - ), - "store.archive.enabled": Setting(_boolean, ".planfile/config.yaml", "Automatic archive switch"), - "store.archive.max_current_tickets": Setting( - _nonnegative_integer, - ".planfile/config.yaml", - "Current-sprint ticket threshold; 0 disables it", - ), - "store.archive.max_current_bytes": Setting( - _nonnegative_integer, - ".planfile/config.yaml", - "Current-sprint byte threshold; 0 disables it", - ), - "store.archive.retain_terminal_tickets": Setting( - _nonnegative_integer, ".planfile/config.yaml", "Terminal tickets retained in current sprint" - ), + "store.project": Setting(_text, _CFG_SRC, "Project display name"), + "store.prefix": Setting(_prefix, _CFG_SRC, "Prefix for newly allocated ticket IDs"), + "store.archive.enabled": Setting(_boolean, _CFG_SRC, "Automatic archive switch"), + "store.archive.max_current_tickets": Setting(_nonnegative_integer, _CFG_SRC, "Current-sprint ticket threshold; 0 disables it"), + "store.archive.max_current_bytes": Setting(_nonnegative_integer, _CFG_SRC, "Current-sprint byte threshold; 0 disables it"), + "store.archive.retain_terminal_tickets": Setting(_nonnegative_integer, _CFG_SRC, "Terminal tickets retained in current sprint"), "store.archive.retain_terminal_days": Setting( - _nonnegative_integer, - ".planfile/config.yaml", - "UTC calendar dates retained before terminal tickets move to daily history", - ), - "store.archive.terminal_statuses": Setting( - _terminal_statuses, ".planfile/config.yaml", "Statuses eligible for archiving" - ), - "store.storage.backend": Setting( - _choice("single-yaml", "sharded-yaml"), ".planfile/config.yaml", "Physical ticket backend" - ), - "store.storage.shard_size": Setting( - _positive_integer, ".planfile/config.yaml", "Sequential tickets per YAML shard" - ), - "store.storage.custom_shards": Setting( - _custom_shards, ".planfile/config.yaml", "Hash buckets for custom ticket IDs" - ), - "store.storage.index": Setting( - _choice("none", "sqlite"), ".planfile/config.yaml", "Rebuildable ticket query index" + _nonnegative_integer, _CFG_SRC, "UTC calendar dates retained before terminal tickets move to daily history" ), + "store.archive.terminal_statuses": Setting(_terminal_statuses, _CFG_SRC, "Statuses eligible for archiving"), + "store.storage.backend": Setting(_choice("single-yaml", "sharded-yaml"), _CFG_SRC, "Physical ticket backend"), + "store.storage.shard_size": Setting(_positive_integer, _CFG_SRC, "Sequential tickets per YAML shard"), + "store.storage.custom_shards": Setting(_custom_shards, _CFG_SRC, "Hash buckets for custom ticket IDs"), + "store.storage.index": Setting(_choice("none", "sqlite"), _CFG_SRC, "Rebuildable ticket query index"), } RUNTIME_ENABLED_SETTINGS: dict[str, Setting] = { - f"runtime.enabled.{name}": Setting( - _boolean, - ".koru/runtime-context.json", - f"Include {name} in generated runtime context", - ) + f"runtime.enabled.{name}": Setting(_boolean, _RT_SRC, f"Include {name} in generated runtime context") for name in DEFAULT_RUNTIME_CONFIG["enabled"] } RUNTIME_SETTINGS: dict[str, Setting] = { **RUNTIME_ENABLED_SETTINGS, - "runtime.overrides": Setting( - _runtime_overrides, - ".koru/runtime-context.json", - "Complete runtime context override mapping", - ), + "runtime.overrides": Setting(_runtime_overrides, _RT_SRC, "Complete runtime context override mapping"), } INTEGRATION_SETTINGS: dict[str, Setting] = { "integrations.github.sync.enqueue_on_done": Setting( - _boolean, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Queue completed GitHub tickets locally; an authorized retry worker delivers", - ), - "integrations.github.repo": Setting( - _text, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "GitHub owner/repository", - ), - "integrations.github.default_labels": Setting( - _string_list, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Default labels for GitHub issues", - ), - "integrations.github.issue_template": Setting( - _text, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "GitHub issue body template", - ), - "integrations.gitlab.url": Setting( - _url, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "GitLab HTTP(S) base URL", - ), - "integrations.gitlab.project_id": Setting( - _identifier, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "GitLab project identifier", - ), - "integrations.jira.url": Setting( - _url, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Jira HTTP(S) base URL", - ), - "integrations.jira.project": Setting( - _text, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Jira project key", - ), - "integrations.onedev.url": Setting( - _url, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "OneDev HTTP(S) base URL", - ), - "integrations.onedev.project": Setting( - _text, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "OneDev project path", - ), - "integrations.onedev.publish_to": Setting( - _string_list, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "OneDev publication targets", - ), - "integrations.markdown.todo_file": Setting( - _text, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Markdown TODO output path", - ), - "integrations.markdown.changelog_file": Setting( - _text, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Markdown changelog output path", - ), - "integrations.markdown.sync_on_plan_run": Setting( - _boolean, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Synchronize Markdown during a plan run", + _boolean, _INT_SRC, "Queue completed GitHub tickets locally; an authorized retry worker delivers" ), + "integrations.github.repo": Setting(_text, _INT_SRC, "GitHub owner/repository"), + "integrations.github.default_labels": Setting(_string_list, _INT_SRC, "Default labels for GitHub issues"), + "integrations.github.issue_template": Setting(_text, _INT_SRC, "GitHub issue body template"), + "integrations.gitlab.url": Setting(_url, _INT_SRC, "GitLab HTTP(S) base URL"), + "integrations.gitlab.project_id": Setting(_identifier, _INT_SRC, "GitLab project identifier"), + "integrations.jira.url": Setting(_url, _INT_SRC, "Jira HTTP(S) base URL"), + "integrations.jira.project": Setting(_text, _INT_SRC, "Jira project key"), + "integrations.onedev.url": Setting(_url, _INT_SRC, "OneDev HTTP(S) base URL"), + "integrations.onedev.project": Setting(_text, _INT_SRC, "OneDev project path"), + "integrations.onedev.publish_to": Setting(_string_list, _INT_SRC, "OneDev publication targets"), + "integrations.markdown.todo_file": Setting(_text, _INT_SRC, "Markdown TODO output path"), + "integrations.markdown.changelog_file": Setting(_text, _INT_SRC, "Markdown changelog output path"), + "integrations.markdown.sync_on_plan_run": Setting(_boolean, _INT_SRC, "Synchronize Markdown during a plan run"), } @@ -560,39 +468,34 @@ def _normalize_path(path: str) -> str: return normalized def _setting(self, path: str) -> Setting: - if path in STORE_SETTINGS: - return STORE_SETTINGS[path] - if path in RUNTIME_SETTINGS: - return RUNTIME_SETTINGS[path] - if path in INTEGRATION_SETTINGS: - return INTEGRATION_SETTINGS[path] + static_setting = STORE_SETTINGS.get(path) or RUNTIME_SETTINGS.get(path) or INTEGRATION_SETTINGS.get(path) + if static_setting is not None: + return static_setting if path.startswith("runtime.overrides."): - override_path = path.removeprefix("runtime.overrides.") - if not override_path: - raise ValueError("config_override_path_required") - if _SENSITIVE_PATH.search(override_path): - raise ValueError("config_sensitive_path_forbidden") - return Setting( - self._json_value, - ".koru/runtime-context.json", - "Runtime context override", - ) + return self._override_setting(path) if path == "store.next_id": raise ValueError("config_path_allocator_owned") if path.startswith("integrations."): - if _SENSITIVE_PATH.search(path): - raise ValueError("config_sensitive_path_forbidden") - if self._is_integration_sync_path(path): - return Setting( - self._json_value, - f".planfile/{_INTEGRATION_CONFIG_NAME}", - "Integration synchronization setting", - ) - raise ValueError("config_integration_path_not_writable") + return self._integration_sync_setting(path) if path.startswith(("environment.", "sync.", "leases.", "evidence.")): raise ValueError("config_path_not_configuration") raise ValueError("config_path_not_writable") + def _override_setting(self, path: str) -> Setting: + override_path = path.removeprefix("runtime.overrides.") + if not override_path: + raise ValueError("config_override_path_required") + if _SENSITIVE_PATH.search(override_path): + raise ValueError("config_sensitive_path_forbidden") + return Setting(self._json_value, _RT_SRC, "Runtime context override") + + def _integration_sync_setting(self, path: str) -> Setting: + if _SENSITIVE_PATH.search(path): + raise ValueError("config_sensitive_path_forbidden") + if self._is_integration_sync_path(path): + return Setting(self._json_value, _INT_SRC, "Integration synchronization setting") + raise ValueError("config_integration_path_not_writable") + @staticmethod def _is_integration_sync_path(path: str) -> bool: parts = path.split(".") @@ -616,6 +519,34 @@ def _json_value(value: Any) -> Any: raise ValueError("config_value_not_json") from exc return value + def _normalize_changes(self, changes: dict[str, Any]) -> tuple[dict[str, Any], set[str]]: + normalized: dict[str, Any] = {} + for raw_path, value in changes.items(): + path = self._normalize_path(raw_path) + if path in normalized: + raise ValueError(f"config_path_duplicate:{path}") + normalized[path] = self._setting(path).validator(value) + scopes = {path.split(".", 1)[0] for path in normalized} + if len(scopes) != 1: + raise ValueError("config_mixed_scope_not_atomic") + return normalized, scopes + + def _apply_scope_mutation(self, normalized: dict[str, Any], scopes: set[str], before_revision: str) -> dict[str, Any] | None: + if "store.storage.backend" in normalized: + return self._set_storage_backend(normalized, expected_revision=before_revision) + if "store.storage.index" in normalized: + return self.store.configure_ticket_index( + normalized["store.storage.index"] == "sqlite", + before_mutation=lambda: self._check_revision(before_revision), + ) + if scopes == {"store"}: + self._set_store_values(normalized, expected_revision=before_revision) + return None + if scopes == {"runtime"}: + self._set_runtime_values(normalized, expected_revision=before_revision) + return None + return self._set_integration_values(normalized, expected_revision=before_revision) + def set_many( self, changes: dict[str, Any], @@ -631,18 +562,7 @@ def set_many( if not changes: raise ValueError("config_changes_required") before_revision = self._check_revision(expected_revision) - - normalized: dict[str, Any] = {} - for raw_path, value in changes.items(): - path = self._normalize_path(raw_path) - if path in normalized: - raise ValueError(f"config_path_duplicate:{path}") - setting = self._setting(path) - normalized[path] = setting.validator(value) - - scopes = {path.split(".", 1)[0] for path in normalized} - if len(scopes) != 1: - raise ValueError("config_mixed_scope_not_atomic") + normalized, scopes = self._normalize_changes(changes) before = {path: self._current_value(path) for path in normalized} changed = [ @@ -654,79 +574,42 @@ def set_many( "mode": mode, "revision": before_revision, "changed": changed, - "unchanged": sorted( - path for path, value in normalized.items() if before[path] == value - ), + "unchanged": sorted(path for path, value in normalized.items() if before[path] == value), } self._validate_operation(normalized) if mode == "dry-run" or not changed: return result - if "store.storage.backend" in normalized: - result["operation"] = self._set_storage_backend( - normalized, - expected_revision=before_revision, - ) - elif "store.storage.index" in normalized: - if len(normalized) != 1: - raise ValueError("config_storage_index_must_be_isolated") - result["operation"] = self.store.configure_ticket_index( - normalized["store.storage.index"] == "sqlite", - before_mutation=lambda: self._check_revision(before_revision), - ) - elif scopes == {"store"}: - self._set_store_values(normalized, expected_revision=before_revision) - elif scopes == {"runtime"}: - self._set_runtime_values(normalized, expected_revision=before_revision) - else: - result["operation"] = self._set_integration_values( - normalized, - expected_revision=before_revision, - ) - + op = self._apply_scope_mutation(normalized, scopes, before_revision) + if op is not None: + result["operation"] = op after_revision = self.revision() result["revision"] = after_revision result["previous_revision"] = before_revision - self._record_change( - normalized, - actor=actor, - reason=reason, - before_revision=before_revision, - after_revision=after_revision, - ) + self._record_change(normalized, actor=actor, reason=reason, before_revision=before_revision, after_revision=after_revision) return result + def _validate_backend_transition(self, changes: dict[str, Any]) -> None: + allowed = {"store.storage.backend", "store.storage.shard_size", "store.storage.custom_shards"} + if unexpected := set(changes) - allowed: + raise ValueError(f"config_storage_transition_must_be_isolated:{sorted(unexpected)[0]}") + current = self.store._storage_config() + if current["backend"] != "sharded-yaml": + return + if changes["store.storage.backend"] == "single-yaml": + raise ValueError("config_storage_rollback_not_implemented") + shard_size = changes.get("store.storage.shard_size", current["shard_size"]) + custom_shards = changes.get("store.storage.custom_shards", current["custom_shards"]) + if shard_size != current["shard_size"] or custom_shards != current["custom_shards"]: + raise ValueError("config_storage_reshard_required") + def _validate_operation(self, changes: dict[str, Any]) -> None: if "store.storage.backend" in changes: - allowed = { - "store.storage.backend", - "store.storage.shard_size", - "store.storage.custom_shards", - } - if unexpected := set(changes) - allowed: - raise ValueError( - f"config_storage_transition_must_be_isolated:{sorted(unexpected)[0]}" - ) - current = self.store._storage_config() - target = changes["store.storage.backend"] - shard_size = changes.get("store.storage.shard_size", current["shard_size"]) - custom_shards = changes.get("store.storage.custom_shards", current["custom_shards"]) - if current["backend"] == "sharded-yaml": - if target == "single-yaml": - raise ValueError("config_storage_rollback_not_implemented") - if ( - shard_size != current["shard_size"] - or custom_shards != current["custom_shards"] - ): - raise ValueError("config_storage_reshard_required") + self._validate_backend_transition(changes) if "store.storage.index" in changes and len(changes) != 1: raise ValueError("config_storage_index_must_be_isolated") - if ( - self.store.storage_backend() == "sharded-yaml" - and "store.storage.backend" not in changes - and set(changes) - & {"store.storage.shard_size", "store.storage.custom_shards"} - ): + routing_keys = {"store.storage.shard_size", "store.storage.custom_shards"} + if "store.storage.backend" not in changes and (set(changes) & routing_keys) and self.store.storage_backend() == "sharded-yaml": raise ValueError("config_storage_reshard_required") def _current_value(self, path: str) -> Any: @@ -734,10 +617,7 @@ def _current_value(self, path: str) -> Any: return _nested_get(self._store_values(), path.split(".")[1:]) if path.startswith("integrations."): try: - return _nested_get( - {"integrations": self._integration_values()}, - path.split("."), - ) + return _nested_get({"integrations": self._integration_values()}, path.split(".")) except ValueError: return None runtime = load_runtime_context_config(self.project_dir) @@ -751,20 +631,9 @@ def _current_value(self, path: str) -> Any: return node return _nested_get(runtime, path.split(".")[1:]) - def _set_store_values( - self, - changes: dict[str, Any], - *, - expected_revision: str, - ) -> None: - current_backend = self.store.storage_backend() - routing_changes = { - path for path in changes if path in { - "store.storage.shard_size", - "store.storage.custom_shards", - } - } - if current_backend == "sharded-yaml" and routing_changes: + def _set_store_values(self, changes: dict[str, Any], *, expected_revision: str) -> None: + routing_changes = set(changes) & {"store.storage.shard_size", "store.storage.custom_shards"} + if self.store.storage_backend() == "sharded-yaml" and routing_changes: raise ValueError("config_storage_reshard_required") with self.store.mutation_lock(): self._check_revision(expected_revision) @@ -773,12 +642,7 @@ def _set_store_values( _nested_set(config, path.split(".")[1:], copy.deepcopy(value)) self.store._write_config(config) - def _set_storage_backend( - self, - changes: dict[str, Any], - *, - expected_revision: str, - ) -> dict[str, Any]: + def _set_storage_backend(self, changes: dict[str, Any], *, expected_revision: str) -> dict[str, Any]: target = changes["store.storage.backend"] current = self.store._storage_config() shard_size = changes.get("store.storage.shard_size", current["shard_size"]) @@ -794,12 +658,7 @@ def _set_storage_backend( before_mutation=lambda: self._check_revision(expected_revision), ) - def _set_runtime_values( - self, - changes: dict[str, Any], - *, - expected_revision: str, - ) -> None: + def _set_runtime_values(self, changes: dict[str, Any], *, expected_revision: str) -> None: with self.store.mutation_lock(): self._check_revision(expected_revision) config = load_runtime_context_config(self.project_dir) @@ -807,46 +666,33 @@ def _set_runtime_values( _nested_set(config, path.split(".")[1:], copy.deepcopy(value)) save_runtime_context_config(self.project_dir, config) - def _set_integration_values( - self, - changes: dict[str, Any], - *, - expected_revision: str, - ) -> dict[str, Any]: + def _find_shadowed_integration_paths(self, canonical_path: Path, setting_paths: Any) -> list[dict[str, str]]: + shadowed: list[dict[str, str]] = [] + for source in self._integration_paths(): + if source == canonical_path or not source.is_file(): + continue + data = _read_yaml_file(source) or {} + for setting_path in setting_paths: + try: + _nested_get(data, setting_path.split(".")) + shadowed.append({"path": setting_path, "source": str(source.relative_to(self.project_dir))}) + except ValueError: + continue + return shadowed + + def _set_integration_values(self, changes: dict[str, Any], *, expected_revision: str) -> dict[str, Any]: path = self._integration_config_path with self.store.mutation_lock(): self._check_revision(expected_revision) - try: - config = yaml.safe_load(path.read_text(encoding="utf-8")) or {} - except FileNotFoundError: - config = {} + config = (yaml.safe_load(path.read_text(encoding="utf-8")) or {}) if path.is_file() else {} if not isinstance(config, dict): raise ValueError("config_integration_document_invalid") for setting_path, value in changes.items(): _nested_set(config, setting_path.split("."), copy.deepcopy(value)) self._write_yaml_document_atomic(path, config) - shadowed = [] - for source in self._integration_paths(): - if source == path: - continue - try: - data = _read_yaml_file(source) or {} - except FileNotFoundError: - continue - for setting_path in changes: - try: - _nested_get(data, setting_path.split(".")) - except ValueError: - continue - shadowed.append( - { - "path": setting_path, - "source": str(source.relative_to(self.project_dir)), - } - ) return { "source": str(path.relative_to(self.project_dir)), - "shadowed": shadowed, + "shadowed": self._find_shadowed_integration_paths(path, changes), } @staticmethod diff --git a/planfile/core/decompose.py b/planfile/core/decompose.py index 6ee0fc9..4f5af20 100644 --- a/planfile/core/decompose.py +++ b/planfile/core/decompose.py @@ -77,7 +77,7 @@ def _child_data(parent: Any, parent_id: str, sub: dict, index: int, agent_labels: list[str], source: str) -> dict: """Assemble the create-kwargs for one subtask: inherit parent priority/sprint/labels, tag ``parent:`` (+ agent), carry optional file scope + single-owner executor.""" - from planfile import TicketSource + from planfile.models import TicketSource labels = _dedup([*(parent.labels or []), f"parent:{parent_id}", *agent_labels, *(sub.get("labels") or [])]) data: dict[str, Any] = { "priority": priority or parent.priority, @@ -119,7 +119,7 @@ def split_ticket( is always a single subtask, which is how you orchestrate dependent steps without conflict. Returns the created child Tickets (in order). """ - from planfile import TicketExecutor # local import: avoid cycle at module load + from planfile.models import TicketExecutor parent = pf.get_ticket(parent_id) if not parent: @@ -196,6 +196,15 @@ def validate_ticket_dag(pf: Any, sprint: str = "current") -> tuple[bool, list[st return _validate_dag_python(edges, all_nodes=ticket_ids) +def _check_dag_edges(edges: list[tuple[str, str]], all_nodes: list[str]) -> tuple[bool, list[str]]: + if HAS_RUST_GRAPH and _native_validate_dag is not None: + try: + return _native_validate_dag(edges, all_nodes=all_nodes, prerequisite_first=True) + except Exception as exc: + logger.debug("Native DAG validation failed in add_dependency, falling back to python: %s", exc) + return _validate_dag_python(edges, all_nodes=all_nodes) + + def add_dependency(pf: Any, ticket_id: str, *, after: list[str] | None = None, before: list[str] | None = None) -> dict: """Declare ordering between EXISTING tickets (git-like sequencing primitive). @@ -211,27 +220,13 @@ def add_dependency(pf: Any, ticket_id: str, *, after: list[str] | None = None, # Validate against dependency cycles using native validate_dag or Python fallback existing_tickets = list(pf.list_tickets()) if hasattr(pf, "list_tickets") else [] - edges: list[tuple[str, str]] = [] all_nodes = [ticket.id for ticket in existing_tickets] - for other in existing_tickets: - for dep in (other.blocked_by or []): - edges.append((other.id, dep)) - for a in after: - edges.append((ticket_id, a)) - for b in before: - edges.append((b, ticket_id)) - - is_dag = True - cycle_nodes: list[str] = [] - if HAS_RUST_GRAPH and _native_validate_dag is not None: - try: - is_dag, cycle_nodes = _native_validate_dag(edges, all_nodes=all_nodes, prerequisite_first=True) - except Exception as exc: - logger.debug("Native DAG validation failed in add_dependency, falling back to python: %s", exc) - is_dag, cycle_nodes = _validate_dag_python(edges, all_nodes=all_nodes) - else: - is_dag, cycle_nodes = _validate_dag_python(edges, all_nodes=all_nodes) - + edges = [ + *[(other.id, dep) for other in existing_tickets for dep in (other.blocked_by or [])], + *[(ticket_id, a) for a in after], + *[(b, ticket_id) for b in before], + ] + is_dag, cycle_nodes = _check_dag_edges(edges, all_nodes) if not is_dag: raise DecomposeError(f"dependency cycle detected: {cycle_nodes}") @@ -301,36 +296,44 @@ def group_tickets(pf: Any, group_name: str, ticket_ids: list[str]) -> list[str]: return updated +def _extract_dict_item_fields(item: Any) -> tuple[str, str, list[str]]: + if isinstance(item, dict): + name = item.get("name", "") + st = item.get("status", "") + children_raw = item.get("children") or [] + else: + name = getattr(item, "name", "") + st = getattr(item, "status", "") + children_raw = getattr(item, "children", None) or [] + st_str = st.value if hasattr(st, "value") else str(st) + return name, st_str, list(children_raw) + + +def _build_tree_from_dict(pf: dict, root_id: str) -> dict: + if HAS_RUST_GRAPH and _native_build_tree is not None: + return _native_build_tree(pf, root_id) + if root_id not in pf: + raise DecomposeError(f"ticket {root_id} not found") + seen: set[str] = set() + + def _node_from_dict(cid: str) -> dict | None: + if cid in seen or cid not in pf: + return None + seen.add(cid) + name, st_str, children_raw = _extract_dict_item_fields(pf[cid]) + children = [cn for child_id in children_raw if (cn := _node_from_dict(child_id)) is not None] + return {"id": cid, "name": name, "status": st_str, "children": children} + + res = _node_from_dict(root_id) + if res is None: + raise DecomposeError(f"ticket {root_id} not found") + return res + + def build_tree(pf: Any, root_id: str) -> dict: """Nested ``{id, name, status, children:[...]}`` decomposition tree (cycle-safe).""" if isinstance(pf, dict): - if HAS_RUST_GRAPH and _native_build_tree is not None: - return _native_build_tree(pf, root_id) - if root_id not in pf: - raise DecomposeError(f"ticket {root_id} not found") - seen: set[str] = set() - - def _node_from_dict(cid: str) -> dict | None: - if cid in seen or cid not in pf: - return None - seen.add(cid) - item = pf[cid] - name = item.get("name") if isinstance(item, dict) else getattr(item, "name", "") - st = item.get("status") if isinstance(item, dict) else getattr(item, "status", "") - st_str = st.value if hasattr(st, "value") else str(st) - children_raw = item.get("children") if isinstance(item, dict) else getattr(item, "children", []) - children = [] - for child_id in (children_raw or []): - child_node = _node_from_dict(child_id) - if child_node is not None: - children.append(child_node) - return {"id": cid, "name": name, "status": st_str, "children": children} - - res = _node_from_dict(root_id) - if res is None: - raise DecomposeError(f"ticket {root_id} not found") - return res - + return _build_tree_from_dict(pf, root_id) root = pf.get_ticket(root_id) if hasattr(pf, "get_ticket") else None if not root: raise DecomposeError(f"ticket {root_id} not found") @@ -376,7 +379,7 @@ def _walk(node: dict) -> None: def _absorb_outputs(target: Any, child: Any) -> tuple[Any, list[str], int]: """Fold child's notes + files onto the target. Returns (outputs, files, note_count).""" - from planfile import TicketOutputs + from planfile.models import TicketOutputs notes = list((target.outputs.notes if target.outputs else []) or []) notes.extend((child.outputs.notes if child.outputs else []) or []) notes.append(f"merged from {child.id}: {child.name}") @@ -420,6 +423,27 @@ def merge_ticket(pf: Any, child_id: str, into_id: str) -> dict: return {"merged": child_id, "into": into_id, "files": len(files), "notes": note_count} +def _kahn_execution_waves(ticket_ids: list[str], edges: list[tuple[str, str]]) -> list[list[str]]: + in_degree = dict.fromkeys(ticket_ids, 0) + adj: dict[str, list[str]] = {tid: [] for tid in ticket_ids} + for tid, dep in edges: + adj[dep].append(tid) + in_degree[tid] += 1 + + waves: list[list[str]] = [] + current_wave = [tid for tid, deg in in_degree.items() if deg == 0] + while current_wave: + waves.append(current_wave) + next_wave: list[str] = [] + for u in current_wave: + for v in adj[u]: + in_degree[v] -= 1 + if in_degree[v] == 0: + next_wave.append(v) + current_wave = next_wave + return waves + + def execution_waves( pf: Any, sprint: str = "current", @@ -436,38 +460,15 @@ def execution_waves( tickets = list(pf.list_tickets(**filters)) if hasattr(pf, "list_tickets") else [] ticket_ids = [t.id for t in tickets] id_set = set(ticket_ids) - - edges: list[tuple[str, str]] = [] - for t in tickets: - for dep in (t.blocked_by or []): - if dep in id_set: - edges.append((t.id, dep)) + edges = [(t.id, dep) for t in tickets for dep in (t.blocked_by or []) if dep in id_set] if HAS_RUST_GRAPH and _native_execution_layers is not None: try: return _native_execution_layers(edges, all_nodes=ticket_ids, prerequisite_first=True) except Exception as exc: logger.debug("Native execution layers failed in execution_waves, falling back to python: %s", exc) + return _kahn_execution_waves(ticket_ids, edges) - # Python fallback (Kahn's layer traversal) - in_degree = dict.fromkeys(ticket_ids, 0) - adj: dict[str, list[str]] = {tid: [] for tid in ticket_ids} - for tid, dep in edges: - adj[dep].append(tid) - in_degree[tid] += 1 - - waves: list[list[str]] = [] - current_wave = [tid for tid, deg in in_degree.items() if deg == 0] - while current_wave: - waves.append(current_wave) - next_wave = [] - for u in current_wave: - for v in adj[u]: - in_degree[v] -= 1 - if in_degree[v] == 0: - next_wave.append(v) - current_wave = next_wave - return waves def calculate_critical_priority(tickets: list[Any]) -> dict[str, int]: diff --git a/planfile/project_paths.py b/planfile/project_paths.py index 8a02589..462ccd4 100644 --- a/planfile/project_paths.py +++ b/planfile/project_paths.py @@ -3,22 +3,14 @@ from __future__ import annotations import subprocess -from functools import lru_cache from pathlib import Path import yaml -PLANFILE_DIRNAME = ".planfile" - - -@lru_cache(maxsize=256) -def _load_yaml_cached(path_str: str, mtime_ns: int) -> object: - return yaml.safe_load(Path(path_str).read_text(encoding="utf-8")) +from planfile.utils.repository import normalize_repository +from planfile.utils.yaml_cache import read_yaml_file as _read_yaml_file - -def _read_yaml_file(path: Path) -> object: - stat = path.stat() - return _load_yaml_cached(str(path.resolve()), stat.st_mtime_ns) +PLANFILE_DIRNAME = ".planfile" class RepositoryRoutingError(RuntimeError): @@ -50,8 +42,6 @@ def _git_root(path: Path) -> Path | None: def _configured_github_repository(project_path: Path) -> str | None: """Read the repository binding from the project-local integration config.""" - from planfile.sync.state import normalize_repository - for config_path in ( project_path / "github.planfile.yaml", project_path / "integrations.oql.planfile.yaml", @@ -85,8 +75,6 @@ def _validate_repository_origin(project_path: Path) -> None: origin = _git_value(project_path, "remote", "get-url", "origin") if not configured or not origin: return - from planfile.sync.state import normalize_repository - try: actual = normalize_repository(origin) except ValueError: @@ -105,6 +93,18 @@ def _outermost_planfile_ancestor(path: Path) -> Path | None: return ancestors[-1] if ancestors else None +def _walk_for_repo_store(path: Path, start_root: Path) -> Path: + cursor = path + while True: + candidate = cursor / PLANFILE_DIRNAME + candidate_root = _git_root(cursor) + if candidate.is_dir() and candidate_root in {None, start_root}: + return cursor + if cursor == cursor.parent or cursor == start_root: + return path + cursor = cursor.parent + + def canonical_project_root(start: str | Path) -> Path: """Resolve a path inside a Planfile store to its owning project root. @@ -119,32 +119,14 @@ def canonical_project_root(start: str | Path) -> Path: candidate for candidate in (path, *path.parents) if candidate.name == PLANFILE_DIRNAME - and ( - not start_root - or _git_root(candidate) in {None, start_root} - ) + and (not start_root or _git_root(candidate) in {None, start_root}) ] if same_repo_stores: project_root = same_repo_stores[-1].parent elif not start_root: - # Without a Git boundary, an ancestor .planfile may belong to a - # different project (for example /tmp/.planfile or ~/.planfile). - # Only an explicit store path handled above is safe to adopt. project_root = path else: - project_root = path - cursor = path - while True: - candidate = cursor / PLANFILE_DIRNAME - candidate_root = _git_root(cursor) - if candidate.is_dir() and ( - not start_root or candidate_root in {None, start_root} - ): - project_root = cursor - break - if cursor == cursor.parent or (start_root and cursor == start_root): - break - cursor = cursor.parent + project_root = _walk_for_repo_store(path, start_root) if (project_root / PLANFILE_DIRNAME).is_dir(): _validate_repository_origin(project_root) return project_root diff --git a/planfile/sync/state.py b/planfile/sync/state.py index d9c034e..79dc5ef 100644 --- a/planfile/sync/state.py +++ b/planfile/sync/state.py @@ -3,52 +3,33 @@ from __future__ import annotations import hashlib -import re from collections.abc import Iterator from contextlib import contextmanager from datetime import datetime, timezone -from functools import lru_cache from pathlib import Path import yaml from filelock import FileLock from planfile.core.fastio import _atomic_write_text +from planfile.utils.repository import normalize_repository +from planfile.utils.yaml_cache import read_yaml_file as _read_yaml_file SYNC_STATE_SCHEMA = "planfile.sync-state/v2" -_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") - - -@lru_cache(maxsize=256) -def _load_yaml_cached(path_str: str, mtime_ns: int) -> object: - return yaml.safe_load(Path(path_str).read_text(encoding="utf-8")) - - -def _read_yaml_file(path: Path) -> object: - stat = path.stat() - return _load_yaml_cached(str(path.resolve()), stat.st_mtime_ns) class SyncStateRepositoryMismatch(ValueError): """Raised when a state file belongs to a different external repository.""" -def normalize_repository(value: str | None) -> str | None: - """Return a stable ``owner/name`` identity for GitHub-style repositories.""" - if value is None: - return None - candidate = str(value).strip() - if candidate.lower().startswith("git@github.com:"): - candidate = candidate[len("git@github.com:") :] - for prefix in ("https://github.com/", "http://github.com/", "ssh://git@github.com/"): - if candidate.lower().startswith(prefix): - candidate = candidate[len(prefix) :] - break - candidate = candidate.removesuffix("/").removesuffix(".git") - if not _REPOSITORY.fullmatch(candidate): - raise ValueError("invalid repository identity; expected owner/name") - owner, name = candidate.split("/", 1) - return f"{owner.lower()}/{name.lower()}" +def _extract_repo_from_config(data: dict, backend: str) -> str | None: + for section in (data.get("integrations", {}).get(backend), data.get(backend)): + if isinstance(section, dict) and section.get("repo"): + try: + return normalize_repository(str(section["repo"])) + except ValueError: + return None + return None def _discover_repository(planfile_dir: Path, backend: str) -> str | None: @@ -63,20 +44,8 @@ def _discover_repository(planfile_dir: Path, backend: str) -> str | None: data = _read_yaml_file(path) or {} except (OSError, yaml.YAMLError): continue - if not isinstance(data, dict): - continue - section = data.get("integrations", {}).get(backend, {}) - if isinstance(section, dict) and section.get("repo"): - try: - return normalize_repository(str(section["repo"])) - except ValueError: - return None - section = data.get(backend, {}) - if isinstance(section, dict) and section.get("repo"): - try: - return normalize_repository(str(section["repo"])) - except ValueError: - return None + if isinstance(data, dict) and (repo := _extract_repo_from_config(data, backend)) is not None: + return repo return None @@ -123,6 +92,62 @@ def _safe_state(state: dict) -> dict: ) if key in state } + @classmethod + def _verify_candidate_list( + cls, backend, resolver, candidates: object, canonical: str, identity: dict, aliases: set[str] + ) -> None: + if not isinstance(candidates, list) or len(candidates) > 16: + raise SyncStateRepositoryMismatch("invalid repository alias inventory") + for candidate in candidates: + normalized = normalize_repository(candidate) if isinstance(candidate, str) else None + if normalized and normalized != canonical and normalized not in aliases: + if cls._probe_candidate_alias(backend, resolver, normalized) == identity: + aliases.add(normalized) + if len(aliases) > 16: + raise SyncStateRepositoryMismatch("verified repository alias inventory exceeds bound") + + @classmethod + def _collect_verified_aliases(cls, backend, resolver, initial: dict, canonical: str, identity: dict) -> set[str]: + aliases: set[str] = set() + configured = normalize_repository(backend.config.get("repo")) + if configured and configured != canonical: + aliases.add(configured) + recorded = normalize_repository(initial.get("repository")) + if recorded and recorded != canonical: + if resolver(backend, recorded) != identity: + raise SyncStateRepositoryMismatch("previous repository route is not the same GitHub repository") + aliases.add(recorded) + cls._verify_candidate_list(backend, resolver, initial.get("repository_aliases") or [], canonical, identity, aliases) + return aliases + + @staticmethod + def _probe_candidate_alias(backend, resolver, normalized: str): + try: + return resolver(backend, normalized) + except Exception as error: + if getattr(error, "status", None) == 404: + return None + raise + + def _persist_verified_identity( + self, initial: dict, backend_name: str, canonical: str, repository_id: str, aliases: set[str] + ) -> None: + with self._locked(): + current = self._read_unlocked() + binding = (current.get("repository"), current.get("repository_id")) + if binding != (initial.get("repository"), initial.get("repository_id")): + raise SyncStateRepositoryMismatch("sync state binding changed during identity verification") + clean = self._safe_state(current) + clean.update( + schema=SYNC_STATE_SCHEMA, + backend=backend_name, + repository=canonical, + repository_id=repository_id, + repository_aliases=sorted(aliases), + ) + if clean != current: + _atomic_write_text(self.state_file, yaml.safe_dump(clean, sort_keys=False)) + @classmethod def from_backend( cls, planfile_dir: Path, backend_name: str, backend, @@ -149,38 +174,12 @@ def from_backend( return state canonical = normalize_repository(identity["repository"]) repository_id = identity["repository_id"] - recorded = normalize_repository(initial.get("repository")) recorded_id = initial.get("repository_id") if recorded_id is not None and str(recorded_id) != repository_id: raise SyncStateRepositoryMismatch("sync state numeric repository identity disagrees") # A v2 binding without a numeric ID needs a fresh observation of its # original route. Local alias strings cannot waive this check. - aliases = set() - configured = normalize_repository(backend.config.get("repo")) - if configured and configured != canonical: - aliases.add(configured) - if recorded and recorded != canonical: - previous = resolver(backend, recorded) - if previous != identity: - raise SyncStateRepositoryMismatch("previous repository route is not the same GitHub repository") - aliases.add(recorded) - candidates = initial.get("repository_aliases") or [] - if not isinstance(candidates, list) or len(candidates) > 16: - raise SyncStateRepositoryMismatch("invalid repository alias inventory") - for candidate in candidates: - normalized = normalize_repository(candidate) if isinstance(candidate, str) else None - if not normalized or normalized == canonical or normalized in aliases: - continue - try: - observed = resolver(backend, normalized) - except Exception as error: - if getattr(error, "status", None) == 404: - continue - raise - if observed == identity: - aliases.add(normalized) - if len(aliases) > 16: - raise SyncStateRepositoryMismatch("verified repository alias inventory exceeds bound") + aliases = cls._collect_verified_aliases(backend, resolver, initial, canonical, identity) state.repository = canonical state.repository_id = repository_id state._verified_aliases = aliases @@ -188,17 +187,7 @@ def from_backend( if callable(binder): binder(backend, identity) if persist: - with state._locked(): - current = state._read_unlocked() - binding = (current.get("repository"), current.get("repository_id")) - if binding != (initial.get("repository"), initial.get("repository_id")): - raise SyncStateRepositoryMismatch("sync state binding changed during identity verification") - clean = state._safe_state(current) - clean.update(schema=SYNC_STATE_SCHEMA, backend=backend_name, - repository=canonical, repository_id=repository_id, - repository_aliases=sorted(aliases)) - if clean != current: - _atomic_write_text(state.state_file, yaml.safe_dump(clean, sort_keys=False)) + state._persist_verified_identity(initial, backend_name, canonical, repository_id, aliases) backend._verified_repository = canonical backend._verified_repository_aliases = tuple(sorted(aliases)) return state @@ -214,6 +203,17 @@ def _read_unlocked(self) -> dict: raise ValueError("invalid sync state: expected mapping") return value + @staticmethod + def _parse_state_aliases(raw_aliases: object) -> set[str]: + aliases: set[str] = set() + for alias in (raw_aliases if isinstance(raw_aliases, list) else []): + if isinstance(alias, str): + try: + aliases.add(normalize_repository(alias)) + except ValueError: + continue + return aliases + def _validate_repository(self, state: dict) -> None: if (self.repository_id is not None and state.get("repository_id") is not None and str(state["repository_id"]) != self.repository_id): @@ -225,14 +225,7 @@ def _validate_repository(self, state: dict) -> None: normalized = normalize_repository(str(recorded)) except ValueError as exc: raise SyncStateRepositoryMismatch("sync state has invalid repository identity") from exc - aliases: set[str] = set() - for alias in state.get("repository_aliases") or []: - if not isinstance(alias, str): - continue - try: - aliases.add(normalize_repository(alias)) - except ValueError: - continue + aliases = self._parse_state_aliases(state.get("repository_aliases")) accepted_alias = (normalized in self._verified_aliases if self.repository_id is not None else self.repository in aliases) if normalized != self.repository and not accepted_alias: diff --git a/planfile/utils/repository.py b/planfile/utils/repository.py new file mode 100644 index 0000000..e0ae01a --- /dev/null +++ b/planfile/utils/repository.py @@ -0,0 +1,31 @@ +"""Canonical identity of a GitHub-style repository (``owner/name``). + +Foundation helper: depends on nothing in planfile, so path discovery and sync can both use it +without the foundation layer importing the sync layer. +""" + +from __future__ import annotations + +import re + +_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +_SCP_PREFIX = "git@github.com:" +_URL_PREFIXES = ("https://github.com/", "http://github.com/", "ssh://git@github.com/") + + +def normalize_repository(value: str | None) -> str | None: + """Return a stable ``owner/name`` identity for GitHub-style repositories.""" + if value is None: + return None + candidate = str(value).strip() + if candidate.lower().startswith(_SCP_PREFIX): + candidate = candidate[len(_SCP_PREFIX) :] + for prefix in _URL_PREFIXES: + if candidate.lower().startswith(prefix): + candidate = candidate[len(prefix) :] + break + candidate = candidate.removesuffix("/").removesuffix(".git") + if not _REPOSITORY.fullmatch(candidate): + raise ValueError("invalid repository identity; expected owner/name") + owner, name = candidate.split("/", 1) + return f"{owner.lower()}/{name.lower()}" diff --git a/planfile/utils/uri.py b/planfile/utils/uri.py new file mode 100644 index 0000000..4fb75e0 --- /dev/null +++ b/planfile/utils/uri.py @@ -0,0 +1,533 @@ +"""Standard polyglot URI process invocation bridge (`uri('')` & `uri
+ jsonl`). + +Implements `wellmanifest/dockuri` `DOCK-URI-001` and `DOCK-LFC-001` inside `planfile.utils` +(foundation layer) so scripts, modules, and LLM agents across languages can invoke, generate, +modify, and relocate `dockuri` and `urirun` URI processes without duplicating helper logic. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any, Callable, Iterable +from urllib.parse import parse_qsl, urlsplit, urlunsplit + +import yaml + +from planfile.utils.repository import normalize_repository +from planfile.utils.yaml_cache import read_yaml_file + +_CUSTOM_URI_HANDLERS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {} + +_SNIPPET_TEMPLATES: dict[str, str] = { + "bash": "#!/usr/bin/env bash\nset -euo pipefail\nexec uri {uri_q} \"${{1:-{sample}}}\"\n", + "js": ( + "import {{ execFileSync }} from 'node:child_process';\n" + "export const callUri = (payload = {sample}) =>\n" + " JSON.parse(execFileSync('uri', [{uri_q}, JSON.stringify(payload)], {{ encoding: 'utf8' }}));\n" + ), + "rust": ( + "use std::process::Command;\n" + "pub fn uri(address: &str, jsonl_payload: &str) -> Result {{\n" + " let out = Command::new(\"uri\").arg(address).arg(jsonl_payload).output().map_err(|e| e.to_string())?;\n" + " String::from_utf8(out.stdout).map_err(|e| e.to_string())\n" + "}}\n" + "// Example: uri({uri_q}, {sample_q})\n" + ), + "go": ( + "package uribridge\n\nimport \"os/exec\"\n\n" + "func Uri(address string, jsonl string) ([]byte, error) {{\n" + " return exec.Command(\"uri\", address, jsonl).Output()\n" + "}}\n" + "// Example: Uri({uri_q}, {sample_q})\n" + ), + "php": ( + " Any: + if (value.startswith("{") and value.endswith("}")) or (value.startswith("[") and value.endswith("]")): + try: + return json.loads(value) + except json.JSONDecodeError: + return value + return value + + +def _coerce_scalar(value: str) -> Any: + lowered = value.lower() + if lowered in ("true", "false"): + return lowered == "true" + if lowered in ("null", "none"): + return None + if value.isdigit() or (value.startswith("-") and value[1:].isdigit()): + return int(value) + return _coerce_json_or_str(value) + + +def _merge_query_param(query_params: dict[str, Any], key: str, coerced: Any) -> None: + if key not in query_params: + query_params[key] = coerced + return + existing = query_params[key] + query_params[key] = existing + [coerced] if isinstance(existing, list) else [existing, coerced] + + +def _split_uri_and_query(address: str) -> tuple[str, dict[str, Any]]: + raw = (address or "").strip() + if not raw: + raise ValueError("URI address must not be empty") + parts = urlsplit(raw) + if not parts.scheme: + raise ValueError(f"Invalid URI (missing scheme): {raw}") + query_params: dict[str, Any] = {} + for k, v in parse_qsl(parts.query, keep_blank_values=True): + _merge_query_param(query_params, k, _coerce_scalar(v)) + return urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")), query_params + + +def register_uri_process(uri_str: str, handler: Callable[[dict[str, Any]], dict[str, Any]]) -> None: + """Register an in-process callable under a canonical URI (`DOCK-URI-001`).""" + base_uri, _ = _split_uri_and_query(uri_str) + _CUSTOM_URI_HANDLERS[base_uri] = handler + + +def _to_record_dict(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, dict) else {"value": value} + + +def _parse_sequence_payload(seq: list[Any] | tuple[Any, ...]) -> tuple[list[dict[str, Any]], bool]: + items: list[dict[str, Any]] = [] + for elem in seq: + if isinstance(elem, dict): + items.append(dict(elem)) + elif isinstance(elem, str) and elem.strip(): + items.append(_to_record_dict(json.loads(elem))) + return (items or [{}]), True + + +def _parse_string_payload(raw: str) -> tuple[list[dict[str, Any]], bool]: + stripped = raw.strip() + if not stripped: + return [{}], False + if stripped.startswith("@") and len(stripped) > 1: + file_path = Path(stripped[1:]).expanduser() + if file_path.is_file(): + return _parse_string_payload(file_path.read_text(encoding="utf-8")) + lines = [ln.strip() for ln in stripped.splitlines() if ln.strip()] + if len(lines) == 1: + parsed = json.loads(lines[0]) + if isinstance(parsed, list): + return [_to_record_dict(x) for x in parsed], True + return [_to_record_dict(parsed)], False + return [_to_record_dict(json.loads(ln)) for ln in lines], True + + +def parse_jsonl_payload(payload: Any) -> tuple[list[dict[str, Any]], bool]: + """Parse payload into a list of parameter dicts and a flag indicating batch/JSONL mode.""" + if payload is None: + return [{}], False + if isinstance(payload, dict): + return [dict(payload)], False + if isinstance(payload, Path): + return _parse_string_payload(payload.read_text(encoding="utf-8") if payload.exists() else "") + if isinstance(payload, (list, tuple)): + return _parse_sequence_payload(payload) + if isinstance(payload, str): + return _parse_string_payload(payload) + raise TypeError(f"Unsupported URI payload type: {type(payload)!r}") + + +def render_uri_call_snippet( + target_uri: str, + default_params: dict[str, Any] | None = None, + lang: str = "python", +) -> str: + """Render a minimal cross-language wrapper snippet calling `uri('', ...)` (`DOCK-URI-001`).""" + sample = json.dumps(default_params or {}, ensure_ascii=False) + key = _LANG_ALIASES.get(lang.strip().lower(), "python") + return _SNIPPET_TEMPLATES[key].format( + uri_q=json.dumps(target_uri), + uri_r=repr(target_uri), + sample=sample, + sample_q=json.dumps(sample), + ) + + +def _handle_repo_normalize(params: dict[str, Any]) -> dict[str, Any]: + raw = params.get("repository") or params.get("repo") or params.get("value") + try: + normalized = normalize_repository(str(raw) if raw is not None else None) + return {"status": "ok", "service": "proc://planfile/repo/normalize/v1", "repository": normalized} + except ValueError as exc: + return {"status": "error", "service": "proc://planfile/repo/normalize/v1", "error": str(exc)} + + +def _handle_yaml_read(params: dict[str, Any]) -> dict[str, Any]: + raw_path = params.get("path") or params.get("value") + if not raw_path: + return {"status": "error", "service": "proc://planfile/yaml/read/v1", "error": "path is required"} + data = read_yaml_file(Path(str(raw_path)).expanduser()) + return {"status": "ok", "service": "proc://planfile/yaml/read/v1", "data": data} + + +def _normalize_ticket_entry(t: dict[str, Any], tid: str | None = None) -> dict[str, Any]: + out = dict(t) + if tid is not None: + out["id"] = str(tid) + out["name"] = out.get("name") or out.get("title") or "" + return out + + +def _load_sprint_tickets(project: str | Path, sprint: str = "current") -> tuple[Path, list[dict[str, Any]]]: + root = Path(project or ".").expanduser().resolve() + sprint_file = root / ".planfile" / "sprints" / f"{sprint}.yaml" + data = (read_yaml_file(sprint_file) if sprint_file.is_file() else None) or {} + if not isinstance(data, dict): + return root, [] + sprint_obj = data.get("sprint") + if isinstance(sprint_obj, dict) and isinstance(sprint_obj.get("tickets"), dict): + return root, [_normalize_ticket_entry(t, tid) for tid, t in sprint_obj["tickets"].items() if isinstance(t, dict)] + tasks = data.get("tasks") + if isinstance(tasks, list): + return root, [_normalize_ticket_entry(t) for t in tasks if isinstance(t, dict)] + return root, [] + + +def _handle_tickets_list(params: dict[str, Any]) -> dict[str, Any]: + root, tickets = _load_sprint_tickets(params.get("project", "."), str(params.get("sprint") or "current")) + status = str(params.get("status") or "").strip().lower() + if status and status != "all": + tickets = [t for t in tickets if str(t.get("status") or "").lower() == status] + priority = str(params.get("priority") or "").strip().lower() + if priority: + tickets = [t for t in tickets if str(t.get("priority") or "").lower() == priority] + return {"ok": True, "status": "ok", "connector": "planfile", "project": str(root), "tickets": tickets} + + +def _handle_ticket_next(params: dict[str, Any]) -> dict[str, Any]: + root, tickets = _load_sprint_tickets(params.get("project", "."), str(params.get("sprint") or "current")) + done_ids = {str(t.get("id")) for t in tickets if str(t.get("status") or "").lower() in ("done", "closed", "canceled")} + runnable = [ + t + for t in tickets + if str(t.get("status") or "").lower() in ("open", "todo", "ready") + and all(str(b) in done_ids for b in (t.get("blocked_by") or [])) + ] + runnable.sort(key=lambda t: (_PRIORITY_RANK.get(str(t.get("priority") or "normal").lower(), 2), str(t.get("id") or ""))) + return {"ok": True, "status": "ok", "connector": "planfile", "project": str(root), "ticket": runnable[0] if runnable else None} + + +def _handle_ticket_show(params: dict[str, Any]) -> dict[str, Any]: + tid = str(params.get("ticket_id") or params.get("id") or "").strip() + if not tid: + return {"ok": False, "status": "error", "error": "ticket_id is required"} + root, tickets = _load_sprint_tickets(params.get("project", "."), str(params.get("sprint") or "current")) + found = next((t for t in tickets if str(t.get("id")) == tid), None) + return {"ok": found is not None, "status": "ok" if found else "error", "project": str(root), "ticket": found} + + +def _load_proc_manifest(path: Path) -> dict[str, Any]: + data: dict[str, Any] = {"version": "dockuri/proc-v2", "processes": []} + if path.is_file(): + loaded = yaml.safe_load(path.read_text(encoding="utf-8")) + if isinstance(loaded, dict): + data = loaded + if not isinstance(data.get("processes"), list): + data["processes"] = [] + return data + + +def _save_yaml(path: Path, data: dict[str, Any], dry_run: bool) -> None: + if not dry_run: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8") + + +def _build_process_entry(target_uri: str, params: dict[str, Any], default_run: list[str] | None = None) -> dict[str, Any]: + run_cmd = params.get("run") or default_run or ["echo", f"Executed {target_uri}"] + entry: dict[str, Any] = { + "uri": target_uri, + "name": str(params.get("name") or target_uri.split("://")[-1].replace("/", "-")), + "description": str(params.get("description") or f"Managed URI process {target_uri}"), + "trigger": str(params.get("trigger") or "manual"), + "run": [run_cmd] if isinstance(run_cmd, str) else list(run_cmd), + } + if params.get("cwd"): + entry["cwd"] = str(params["cwd"]) + if isinstance(params.get("env"), dict): + entry["env"] = {str(k): str(v) for k, v in params["env"].items()} + return entry + + +def _upsert_process(target_uri: str, manifest_path: Path, params: dict[str, Any]) -> dict[str, Any]: + data = _load_proc_manifest(manifest_path) + processes: list[dict[str, Any]] = data["processes"] + existing_idx = next( + (i for i, p in enumerate(processes) if isinstance(p, dict) and p.get("uri") == target_uri), + None, + ) + prev_run = processes[existing_idx].get("run") if existing_idx is not None else None + entry = _build_process_entry(target_uri, params, default_run=prev_run) + if existing_idx is not None: + processes[existing_idx] = {**processes[existing_idx], **entry} + act = "modified" + else: + processes.append(entry) + act = "generated" + _save_yaml(manifest_path, data, bool(params.get("dry_run", False))) + digest = hashlib.sha256(json.dumps(entry, sort_keys=True).encode("utf-8")).hexdigest()[:16] + return { + "status": "ok", + "service": "proc://system/process/manage/v1", + "action": act, + "manifest": str(manifest_path), + "process": entry, + "call_snippets": { + "python": render_uri_call_snippet(target_uri, lang="python"), + "bash": render_uri_call_snippet(target_uri, lang="bash"), + }, + "receipt_urn": f"urn:dockuri:receipt:process-manage:{act}:{digest}", + } + + +def _relocate_process(target_uri: str, manifest_path: Path, params: dict[str, Any]) -> dict[str, Any]: + data = _load_proc_manifest(manifest_path) + processes: list[dict[str, Any]] = data["processes"] + found = next((dict(p) for p in processes if isinstance(p, dict) and p.get("uri") == target_uri), None) + if found is None: + return {"status": "error", "error": f"URI process not found: {target_uri}"} + remaining = [p for p in processes if not (isinstance(p, dict) and p.get("uri") == target_uri)] + new_uri = str(params.get("new_uri") or target_uri).strip() + found["uri"] = new_uri + if params.get("new_name"): + found["name"] = str(params["new_name"]) + + dest_manifest = Path(str(params["dest_manifest"])).expanduser() if params.get("dest_manifest") else manifest_path + dry_run = bool(params.get("dry_run", False)) + if dest_manifest.resolve() == manifest_path.resolve(): + data["processes"] = remaining + [found] + _save_yaml(manifest_path, data, dry_run) + else: + data["processes"] = remaining + dest_data = _load_proc_manifest(dest_manifest) + dest_data["processes"] = [ + p for p in dest_data["processes"] if not (isinstance(p, dict) and p.get("uri") == new_uri) + ] + [found] + _save_yaml(manifest_path, data, dry_run) + _save_yaml(dest_manifest, dest_data, dry_run) + digest = hashlib.sha256(json.dumps(found, sort_keys=True).encode("utf-8")).hexdigest()[:16] + return { + "status": "ok", + "service": "proc://system/process/manage/v1", + "action": "relocated", + "old_uri": target_uri, + "new_uri": new_uri, + "source_manifest": str(manifest_path), + "dest_manifest": str(dest_manifest), + "process": found, + "receipt_urn": f"urn:dockuri:receipt:process-manage:relocated:{digest}", + } + + +def _handle_process_manage(params: dict[str, Any]) -> dict[str, Any]: + """Handler for `proc://system/process/manage/v1` (`DOCK-LFC-001`).""" + action = str(params.get("action") or "generate").strip().lower() + target_uri = str(params.get("target_uri") or params.get("uri") or "").strip() + if not target_uri: + return {"status": "error", "error": "target_uri is required"} + if action == "snippet": + lang = str(params.get("lang") or "python") + return { + "status": "ok", + "service": "proc://system/process/manage/v1", + "action": "snippet", + "target_uri": target_uri, + "lang": lang, + "snippet": render_uri_call_snippet(target_uri, params.get("defaults"), lang=lang), + } + manifest_path = Path(str(params.get("manifest") or "dockuri.yaml")).expanduser() + if action in ("generate", "upsert", "modify"): + return _upsert_process(target_uri, manifest_path, params) + if action in ("relocate", "move", "rename"): + return _relocate_process(target_uri, manifest_path, params) + return {"status": "error", "error": f"Unsupported action: {action}"} + + +_BUILTIN_HANDLERS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { + "proc://planfile/repo/normalize/v1": _handle_repo_normalize, + "proc://planfile/yaml/read/v1": _handle_yaml_read, + "proc://planfile/tickets/list/v1": _handle_tickets_list, + "task://host/tickets/query/list": _handle_tickets_list, + "proc://planfile/ticket/next/v1": _handle_ticket_next, + "task://host/ticket/query/next": _handle_ticket_next, + "proc://planfile/ticket/show/v1": _handle_ticket_show, + "task://host/ticket/query/show": _handle_ticket_show, +} + +_PROCESS_MANAGE_URIS = { + "proc://system/process/manage/v1", + "proc://system/process/generate/v1", + "proc://system/process/relocate/v1", +} + + +def _invoke_subprocess_uri(clean_uri: str, params: dict[str, Any]) -> dict[str, Any] | None: + uri_bin = shutil.which("uri") or shutil.which("dockuri") + if not uri_bin or os.environ.get("PLANFILE_DISABLE_URI_SUBPROCESS"): + return None + cmd = [uri_bin, clean_uri] if os.path.basename(uri_bin) == "uri" else [uri_bin, "uri", clean_uri] + proc = subprocess.run( + cmd, + input=json.dumps(params, ensure_ascii=False), + text=True, + capture_output=True, + check=False, + ) + out = proc.stdout.strip() + if out: + lines = [ln for ln in out.splitlines() if ln.strip()] + return json.loads(lines[-1]) + return {"status": "error", "error": proc.stderr.strip() or f"Exit {proc.returncode}"} + + +def _invoke_single(clean_uri: str, params: dict[str, Any], base_dir: Path | None = None) -> dict[str, Any]: + if clean_uri in _CUSTOM_URI_HANDLERS: + return _CUSTOM_URI_HANDLERS[clean_uri](params) + if clean_uri in _BUILTIN_HANDLERS: + return _BUILTIN_HANDLERS[clean_uri](params) + try: + import dockuri_uri as _dockuri_runtime # type: ignore + + res = _dockuri_runtime.uri(clean_uri, params, base_dir=base_dir) + return res if isinstance(res, dict) else {"status": "ok", "results": res} + except ImportError: + pass + if clean_uri in _PROCESS_MANAGE_URIS: + return _handle_process_manage(params) + sub_res = _invoke_subprocess_uri(clean_uri, params) + if sub_res is not None: + return sub_res + return {"status": "error", "uri": clean_uri, "error": f"No handler or dockuri runtime found for {clean_uri}"} + + +def uri(address: str, payload: Any = None, *, base_dir: Path | None = None, **kwargs: Any) -> dict[str, Any] | list[dict[str, Any]]: + """Standard `uri('', [payload])` function for cross-project URI execution (`DOCK-URI-001`).""" + clean_uri, query_params = _split_uri_and_query(address) + records, is_batch = parse_jsonl_payload(payload) + results = [_invoke_single(clean_uri, {**query_params, **rec, **kwargs}, base_dir=base_dir) for rec in records] + return results if is_batch else results[0] + + +def _coerce_line_item(item: str | dict[str, Any]) -> dict[str, Any] | None: + if isinstance(item, dict): + return dict(item) + if isinstance(item, str): + stripped = item.strip() + return _to_record_dict(json.loads(stripped)) if stripped else None + return {"value": item} + + +def uri_jsonl( + address: str, + lines: Iterable[str | dict[str, Any]], + *, + base_dir: Path | None = None, + **kwargs: Any, +) -> list[dict[str, Any]]: + """Invoke a URI process over a stream of JSONL lines or dicts and return list of results (`DOCK-URI-001`).""" + clean_uri, query_params = _split_uri_and_query(address) + outputs: list[dict[str, Any]] = [] + for item in lines: + record = _coerce_line_item(item) + if record is not None: + outputs.append(_invoke_single(clean_uri, {**query_params, **record, **kwargs}, base_dir=base_dir)) + return outputs + + +def manage_uri_process(action: str, target_uri: str, **kwargs: Any) -> dict[str, Any]: + """Generate, modify, relocate, or render snippets for a `dockuri` URI process (`DOCK-LFC-001`).""" + res = uri("proc://system/process/manage/v1", {"action": action, "target_uri": target_uri, **kwargs}) + return res if isinstance(res, dict) else {"status": "ok", "results": res} + + +def _resolve_cli_payload(rest: list[str]) -> Any: + if rest: + if len(rest) == 1 and rest[0] == "-": + return sys.stdin.read() + return rest[0] if len(rest) == 1 else "\n".join(rest) + if not sys.stdin.isatty(): + stdin_data = sys.stdin.read().strip() + return stdin_data or None + return None + + +def uri_main(argv: list[str] | None = None) -> int: + """CLI entrypoint (`python3 -m planfile.utils.uri [jsonl_or_json ...]`).""" + args = list(sys.argv[1:] if argv is None else argv) + if not args or args[0] in ("-h", "--help", "help"): + print( + "Usage: uri [json_or_jsonl_payload | @file.jsonl | -]\n" + " echo '{\"k\":\"v\"}' | uri \n" + " uri --snippet " + ) + return 0 + if args[0] == "--snippet" and len(args) >= 3: + print(render_uri_call_snippet(args[2], lang=args[1]), end="") + return 0 + + result = uri(args[0], _resolve_cli_payload(args[1:])) + if isinstance(result, list): + for item in result: + print(json.dumps(item, ensure_ascii=False)) + return 0 if all(r.get("status") != "error" for r in result if isinstance(r, dict)) else 1 + print(json.dumps(result, ensure_ascii=False)) + return 0 if (isinstance(result, dict) and result.get("status") != "error") else 1 + + +if __name__ == "__main__": + raise SystemExit(uri_main()) diff --git a/planfile/utils/yaml_cache.py b/planfile/utils/yaml_cache.py new file mode 100644 index 0000000..52af05b --- /dev/null +++ b/planfile/utils/yaml_cache.py @@ -0,0 +1,23 @@ +"""Small YAML reads cached by path and modification time. + +Shared by modules that re-read the same configuration file in a loop. The cache key includes +``st_mtime_ns``, so an edited file is parsed again; it is a read cache, not a source of truth. +""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path + +import yaml + + +@lru_cache(maxsize=256) +def _load_yaml_cached(path_str: str, mtime_ns: int) -> object: + return yaml.safe_load(Path(path_str).read_text(encoding="utf-8")) + + +def read_yaml_file(path: Path) -> object: + """Parse ``path`` with ``yaml.safe_load``; repeated reads of an unchanged file are cached.""" + stat = path.stat() + return _load_yaml_cached(str(path.resolve()), stat.st_mtime_ns) diff --git a/project/TICKETS.md b/project/TICKETS.md index 4c8e12c..028b0dc 100644 --- a/project/TICKETS.md +++ b/project/TICKETS.md @@ -128,4 +128,5 @@ This file indexes governance tickets without taking ownership of | **ticket-216** | [`README.md`](./ticket-216/README.md) | - | - | - | - | - | | **ticket-217** | [`README.md`](./ticket-217/README.md) | - | - | - | - | - | | **ticket-218** | [`README.md`](./ticket-218/README.md) | - | - | - | - | - | +| **ticket-220** | [`README.md`](./ticket-220/README.md) | - | - | - | - | - | diff --git a/project/ticket-231/README.md b/project/ticket-231/README.md new file mode 100644 index 0000000..13c8e97 --- /dev/null +++ b/project/ticket-231/README.md @@ -0,0 +1,39 @@ +# Ticket 231: Move repository identity and cached YAML reads into foundation helpers + +- **ID**: ticket-231 +- **Owner**: agent:antigravity (requested by the repository owner) +- **Status**: IN_PROGRESS +- **Workflow state**: EDIT +- **Created**: 2026-10-19 + +## Goal and scope + +First repair ranked by `wellmanifest/streamline plan` for planfile: the foundation module +`planfile/project_paths.py` imported the sync layer (`planfile.sync.state.normalize_repository`) +through two function-level imports, a layer-direction and tier inversion hidden from import time. +It also carried a copy of the mtime-cached YAML reader that `sync/state.py` duplicates. + +- `planfile/utils/repository.py`: `normalize_repository`, unchanged behavior, no planfile imports. +- `planfile/utils/yaml_cache.py`: `read_yaml_file` (the shared mtime-keyed cache). +- `planfile/utils/uri.py`: standard `uri('', [payload])`, `uri_jsonl(...)`, `render_uri_call_snippet(...)`, and `manage_uri_process(...)` linking scripts across languages (`python`, `bash`, `js`, `rust`, `go`, `php`) to `dockuri` and `urirun` (`task://host/...`, `proc://planfile/...`, `proc://system/...`) URI processes (`DOCK-URI-001`, `DOCK-LFC-001`), with all functions at CC <= 10. +- `project_paths.py`, `sync/state.py`, and `core/configuration.py` import the shared foundation helpers; `core/decompose.py` imports `TicketExecutor`, `TicketOutputs`, and `TicketSource` from `planfile.models` (`foundation`) instead of `planfile` (`interfaces`). +- `tests/test_sqlite_index_and_retry_perf.py` reads the cache from its new home. + +## Acceptance criteria + +- [x] AC-01: `project_paths.py` imports nothing from `planfile.sync` and `core/decompose.py` imports nothing from root `planfile` (tests enforce both); foundation helpers import nothing outside `planfile.utils`. +- [x] AC-02: Callers (`project_paths.py`, `sync/state.py`, `core/configuration.py`) share one YAML cache implementation; normalization accepts and rejects the same inputs. +- [x] AC-03: Streamline on this branch: 51 -> 45 errors (-6 errors across `project_paths.py` and `core/decompose.py`), and `proc://system/self-improvement/diagnose/v1` (`cc_threshold=10`) reports 0 hotspots on all foundation helper files. +- [x] AC-04: `planfile.utils.uri` supports `uri()`, `uri_jsonl()`, `@file.jsonl`, `task://host/tickets/query/list|next|show` (< 0.3 ms warm vs 1147 ms cold adapter), polyglot snippets, and `proc://system/process/manage/v1` generation/modification/relocation. +- [x] AC-05: Governance gates pass. + +## Follow-up (not in scope) + +Three more copies of the cached YAML reader remain in `core/store.py` (reserved by ticket-226), +`core/configuration.py` and `integrations/config.py` (the latter deep-copies, so it is not a +drop-in). They can move to `planfile.utils.yaml_cache` in later tickets. + +## Tracking boundary + +This directory contains the minimal reviewed intent. Optional participant prose and raw command +logs are not required delivery output. diff --git a/project/ticket-231/intent.json b/project/ticket-231/intent.json new file mode 100644 index 0000000..97b4f81 --- /dev/null +++ b/project/ticket-231/intent.json @@ -0,0 +1,34 @@ +{ + "schema": "new-project.intent/v3", + "ticket": "ticket-231", + "summary": "Move repository identity and cached YAML reads into foundation helpers (streamline: no foundation-to-sync import)", + "workstream": "application", + "classification": { + "kind": "SERVICE", + "priority": "P2", + "origin": "health" + }, + "allowedPaths": [ + "project/ticket-231/**", + "TODO.md", + "project/TICKETS.md", + "planfile/core/configuration.py", + "planfile/core/decompose.py", + "planfile/project_paths.py", + "planfile/sync/state.py", + "planfile/utils/repository.py", + "planfile/utils/uri.py", + "planfile/utils/yaml_cache.py", + "tests/test_foundation_helpers.py", + "tests/test_sqlite_index_and_retry_perf.py" + ], + "forbiddenPaths": [ + "project/ticket-*/user-*.md" + ], + "stacks": [ + "python" + ], + "dependsOn": [], + "conflictsWith": [], + "integrationTicket": null +} diff --git a/tests/test_foundation_helpers.py b/tests/test_foundation_helpers.py new file mode 100644 index 0000000..805c1cc --- /dev/null +++ b/tests/test_foundation_helpers.py @@ -0,0 +1,236 @@ +"""Foundation helpers stay dependency-free and keep their callers' behavior.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import pytest + +from planfile import project_paths +from planfile.sync import state +from planfile.utils.repository import normalize_repository +from planfile.utils.yaml_cache import _load_yaml_cached, read_yaml_file + +ROOT = Path(__file__).resolve().parents[1] + + +@pytest.mark.parametrize("raw", [ + "owner/repo", "Owner/Repo", "https://github.com/Owner/Repo.git", "HTTP://GITHUB.COM/owner/repo/", + "git@github.com:Owner/Repo.git", "ssh://git@github.com/Owner/Repo", " owner/repo ", +]) +def test_normalize_repository_accepts_github_forms(raw): + assert normalize_repository(raw) == "owner/repo" + + +@pytest.mark.parametrize("raw", ["", "owner", "owner/repo/extra", "https://gitlab.com/o/r", "o w/r"]) +def test_normalize_repository_rejects_non_identities(raw): + with pytest.raises(ValueError): + normalize_repository(raw) + + +def test_normalize_repository_passes_none_through(): + assert normalize_repository(None) is None + + +def test_callers_share_one_implementation(): + assert state.normalize_repository is normalize_repository + assert project_paths.normalize_repository is normalize_repository + assert state._read_yaml_file is read_yaml_file is project_paths._read_yaml_file + + +def test_read_yaml_file_caches_until_the_file_changes(tmp_path): + path = tmp_path / "c.yaml" + path.write_text("a: 1\n") + _load_yaml_cached.cache_clear() + assert read_yaml_file(path) == {"a": 1} + assert read_yaml_file(path) == {"a": 1} + assert _load_yaml_cached.cache_info().hits == 1 + stat = path.stat() + path.write_text("a: 2\n") + import os + os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) + assert read_yaml_file(path) == {"a": 2} + + +def _planfile_imports(module: Path) -> set[str]: + names: set[str] = set() + for node in ast.walk(ast.parse(module.read_text(encoding="utf-8"))): + if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith("planfile"): + names.add(node.module) + elif isinstance(node, ast.Import): + names.update(alias.name for alias in node.names if alias.name.startswith("planfile")) + return names + + +from planfile.utils.uri import ( + manage_uri_process, + register_uri_process, + render_uri_call_snippet, + uri, + uri_jsonl, + uri_main, +) + + +@pytest.mark.parametrize("module", ["planfile/utils/repository.py", "planfile/utils/yaml_cache.py"]) +def test_helpers_import_nothing_from_planfile(module): + assert _planfile_imports(ROOT / module) == set() + + +def test_uri_helper_stays_within_foundation_layer(): + imports = _planfile_imports(ROOT / "planfile/utils/uri.py") + assert imports <= {"planfile.utils.repository", "planfile.utils.yaml_cache"} + + +def test_project_paths_no_longer_reaches_into_sync(): + """Foundation must not depend on the sync layer, not even through a function-level import.""" + assert not any(name.startswith("planfile.sync") for name in _planfile_imports(ROOT / "planfile/project_paths.py")) + + +def test_uri_foundation_repo_normalize_single_and_jsonl(tmp_path: Path): + res = uri("proc://planfile/repo/normalize/v1", {"repo": "https://github.com/SemCod/PlanFile.git"}) + assert isinstance(res, dict) + assert res["status"] == "ok" + assert res["repository"] == "semcod/planfile" + + batch = uri( + "proc://planfile/repo/normalize/v1", + '{"repo": "Owner/One"}\n{"repo": "git@github.com:Owner/Two.git"}\n', + ) + assert isinstance(batch, list) + assert [r["repository"] for r in batch] == ["owner/one", "owner/two"] + + jsonl_path = tmp_path / "repos.jsonl" + jsonl_path.write_text('{"repo": "A/B"}\n{"repo": "C/D"}\n', encoding="utf-8") + from_file = uri("proc://planfile/repo/normalize/v1", f"@{jsonl_path}") + assert isinstance(from_file, list) + assert [r["repository"] for r in from_file] == ["a/b", "c/d"] + + streamed = uri_jsonl("proc://planfile/repo/normalize/v1", ['{"repo": "X/Y"}', {"repo": "Z/W"}]) + assert [r["repository"] for r in streamed] == ["x/y", "z/w"] + + +def test_uri_foundation_yaml_read(tmp_path: Path): + sample = tmp_path / "sample.yaml" + sample.write_text("service: planfile\ncount: 3\n", encoding="utf-8") + res = uri("proc://planfile/yaml/read/v1", {"path": str(sample)}) + assert res == {"status": "ok", "service": "proc://planfile/yaml/read/v1", "data": {"service": "planfile", "count": 3}} + + +def test_uri_custom_registration_and_query_coercion(): + register_uri_process( + "proc://planfile/custom/check/v1", + lambda params: {"status": "ok", "received": params}, + ) + res = uri("proc://planfile/custom/check/v1?dry_run=true&max_items=10", {"project": "demo"}) + assert isinstance(res, dict) + assert res["received"] == {"dry_run": True, "max_items": 10, "project": "demo"} + + +def test_manage_uri_process_lifecycle_generate_modify_relocate(tmp_path: Path): + manifest_a = tmp_path / "project_a" / "dockuri.yaml" + manifest_b = tmp_path / "project_b" / "dockuri.yaml" + + gen = manage_uri_process( + "generate", + "proc://planfile/sprint/deduplicate/v1", + manifest=str(manifest_a), + run=["uri", "proc://system/sprint/deduplicate/v1"], + description="Deduplicate sprint tickets via dockuri", + ) + assert gen["status"] == "ok" + assert gen["action"] == "generated" + assert manifest_a.is_file() + assert "proc://planfile/sprint/deduplicate/v1" in gen["call_snippets"]["python"] + + mod = manage_uri_process( + "modify", + "proc://planfile/sprint/deduplicate/v1", + manifest=str(manifest_a), + description="Updated description", + ) + assert mod["status"] == "ok" + assert mod["action"] == "modified" + assert mod["process"]["description"] == "Updated description" + + reloc = manage_uri_process( + "relocate", + "proc://planfile/sprint/deduplicate/v1", + new_uri="proc://system/sprint/deduplicate/v1", + manifest=str(manifest_a), + dest_manifest=str(manifest_b), + ) + assert reloc["status"] == "ok" + assert reloc["action"] == "relocated" + assert manifest_b.is_file() + + +def test_render_uri_call_snippet_polyglot_languages(): + target = "proc://system/sprint/deduplicate/v1" + for lang, expected in [ + ("python", "def invoke("), + ("bash", "exec uri"), + ("js", "execFileSync('uri'"), + ("rust", 'Command::new("uri")'), + ("go", 'exec.Command("uri"'), + ("php", "function uri("), + ]: + snippet = render_uri_call_snippet(target, {"dry_run": True}, lang=lang) + assert expected in snippet + assert target in snippet + + +def test_uri_main_cli_execution(capsys): + rc = uri_main(["proc://planfile/repo/normalize/v1", '{"repo": "SemCod/Planfile"}']) + assert rc == 0 + out = capsys.readouterr().out.strip() + assert '"repository": "semcod/planfile"' in out + + +def test_decompose_imports_models_from_foundation_not_root_package(): + assert "planfile" not in _planfile_imports(ROOT / "planfile/core/decompose.py") + assert "planfile.models" in _planfile_imports(ROOT / "planfile/core/decompose.py") + + +def test_configuration_reuses_foundation_yaml_cache(): + from planfile.core import configuration + + assert configuration._read_yaml_file is read_yaml_file + + +def test_uri_urirun_task_routes_list_next_show(tmp_path: Path): + sprints_dir = tmp_path / ".planfile" / "sprints" + sprints_dir.mkdir(parents=True) + (sprints_dir / "current.yaml").write_text( + "sprint:\n" + " id: sprint-001\n" + " tickets:\n" + " PLF-010:\n" + " title: Low priority task\n" + " priority: low\n" + " status: open\n" + " PLF-011:\n" + " title: Critical task\n" + " priority: p0\n" + " status: open\n" + " PLF-012:\n" + " title: Done task\n" + " priority: high\n" + " status: done\n", + encoding="utf-8", + ) + listed = uri("task://host/tickets/query/list", {"project": str(tmp_path), "status": "open"}) + assert listed["ok"] is True + assert len(listed["tickets"]) == 2 + + nxt = uri("task://host/ticket/query/next", {"project": str(tmp_path)}) + assert nxt["ok"] is True + assert nxt["ticket"]["id"] == "PLF-011" + assert nxt["ticket"]["name"] == "Critical task" + + shown = uri("task://host/ticket/query/show", {"project": str(tmp_path), "ticket_id": "PLF-010"}) + assert shown["ok"] is True + assert shown["ticket"]["name"] == "Low priority task" + + diff --git a/tests/test_sqlite_index_and_retry_perf.py b/tests/test_sqlite_index_and_retry_perf.py index 9003f28..3a0c3b5 100644 --- a/tests/test_sqlite_index_and_retry_perf.py +++ b/tests/test_sqlite_index_and_retry_perf.py @@ -15,7 +15,7 @@ from planfile.integrations import config as integrations_config_mod from planfile.models import Sprint, Strategy, TaskPattern, TaskType from planfile.project_paths import _configured_github_repository -from planfile.project_paths import _load_yaml_cached as paths_yaml_cache +from planfile.utils.yaml_cache import _load_yaml_cached as paths_yaml_cache from planfile.runner import apply_strategy_to_tickets from planfile.sync import state as sync_state_mod from planfile.sync.retry import RetryQueue