From 56f44ab23c2c998f5cc104b0f2ac6fa633c36a3b Mon Sep 17 00:00:00 2001 From: Ruben Cuevas Date: Fri, 14 Aug 2026 22:16:55 -0400 Subject: [PATCH] feat: auto-apply prose-only skill updates when approval is disabled --- .../modules/memory-skills-hooks.md | 2 +- src/kiro_crew/config/loader.py | 6 +- src/kiro_crew/dashboard/server.py | 55 +- src/kiro_crew/history.py | 185 +++- src/kiro_crew/skills.py | 334 ++++++- test/test_skill_update_auto_apply.py | 863 ++++++++++++++++++ test/test_skill_update_flow.py | 227 ++++- 7 files changed, 1653 insertions(+), 19 deletions(-) create mode 100644 test/test_skill_update_auto_apply.py diff --git a/docs/system-specs/modules/memory-skills-hooks.md b/docs/system-specs/modules/memory-skills-hooks.md index f614162196b..f45d259011e 100644 --- a/docs/system-specs/modules/memory-skills-hooks.md +++ b/docs/system-specs/modules/memory-skills-hooks.md @@ -1015,7 +1015,7 @@ imports deduplicate through the provenance ledger. Hermes-style autonomous skill creation from completed sessions. **Opt-in, and STAGED for approval** — generation is **off by default** (`skills.auto_create_from_sessions` defaults **false**; enable via `kirocrew config set skills.auto_create_from_sessions true` or dashboard Settings → Skills). When on, candidates land in a pending-approval queue (`skills.approval_required` defaults **true**) and nothing goes live unattended. Pipeline: detect (during consolidation) → generate → metadata dedupe → pending queue → human approval → live → archive-if-unused. Key v2 elements (all under `skills.*`): -- **Staged approval:** new skills route to `auto/.pending//`; approve promotes to `auto//` (dashboard: Skills → Pending review). Auto-approve for prose-only is opt-in via `approval_required=false`; **script-bearing candidates always require approval**. +- **Staged approval:** new skills route to `auto/.pending//`; approve promotes to `auto//` (dashboard: Skills → Pending review). Auto-approve for prose-only is opt-in via `approval_required=false` — this covers both new candidates and prose-only **updates** to live auto-skills (an auto-applied update snapshots the prior version to `.versions/` for rollback; a failed auto-apply falls back to the pending queue); **script-bearing candidates and updates always require approval**. - **Scripts:** deterministic procedures may ship a validated **Python** helper (`generate_scripts`, default true); statically validated (regex denylist + AST policy: no dynamic exec/import, destructive fs, process exec, network egress, ≤4 KB) and re-validated at the approve choke point. - **Bounding:** archive-not-delete lifecycle `active→stale(`stale_after_days`,30)→archived(`archive_after_days`,90)`, `max_auto_skills` (100) backstop, pin + cron-referenced exemptions, never-used grace floor; pending TTL `pending_ttl_days` (30). - **Dedupe:** embedding-free metadata comparison over all generated skills (`judge_model`). diff --git a/src/kiro_crew/config/loader.py b/src/kiro_crew/config/loader.py index 23e573184f2..434c09999c2 100644 --- a/src/kiro_crew/config/loader.py +++ b/src/kiro_crew/config/loader.py @@ -3044,8 +3044,10 @@ class SkillsConfig: metadata=_meta( "Skill Approval Required", "When true, auto-generated skill candidates land in a pending queue for " - "human review instead of going live. Prose-only skills may auto-publish " - "when this is false; skills that bundle scripts ALWAYS require approval " + "human review instead of going live. When false, prose-only NEW skills " + "auto-publish and prose-only UPDATES to live auto-skills auto-apply " + "(the prior version is snapshotted under .versions/ for rollback); " + "skills or updates that bundle scripts ALWAYS require approval " "regardless of this flag.", ), ) diff --git a/src/kiro_crew/dashboard/server.py b/src/kiro_crew/dashboard/server.py index 88d8364edf7..b9fcb239ccc 100644 --- a/src/kiro_crew/dashboard/server.py +++ b/src/kiro_crew/dashboard/server.py @@ -159,7 +159,12 @@ from kiro_crew.security import redact_credentials, redact_exfiltration_urls from kiro_crew.sel import sel from kiro_crew.skill_usage import register_skill_read_observer -from kiro_crew.skills import SkillsLoader, set_pending_consumed_hook, set_pending_staged_hook +from kiro_crew.skills import ( + SkillsLoader, + set_pending_consumed_hook, + set_pending_staged_hook, + set_update_auto_applied_hook, +) from kiro_crew.tunnel.setup import setup_tunnel if TYPE_CHECKING: @@ -2196,6 +2201,54 @@ def _resolve() -> None: logger.debug("pending-skill notification resolve failed", exc_info=True) set_pending_consumed_hook(_on_pending_skill_consumed) + + def _on_update_auto_applied(info: dict) -> None: + # Informational only — the update is already live (approval is + # disabled), so this must read as "went live", never as a review + # request. Kept minimal: identity, new version, and the fact that + # the prior version is restorable. + try: + target = str(info.get("target") or info.get("name") or "skill") + slug = str(info.get("slug") or "") + version = info.get("new_version") + description = str(info.get("description") or "").strip() + title = "Skill auto-updated" + head = f"**{target}** auto-updated to v{version}." + if description: + head = f"**{target}** — {description}\n\nAuto-updated to v{version}." + body = ( + head + + "\n\nApplied without review because skill approval is disabled; " + "the previous version was snapshotted and can be restored from " + "the skill's version history." + ) + payload = { + "slug": slug, + "target": target, + "new_version": version, + } + skills_url = "/capabilities?tab=skills" + + def _emit() -> None: + try: + state.notify("skills", title, body, meta=payload, url=skills_url) + # The candidate left the pending queue; nudge open + # dashboards to refresh it. + state.broadcast_ws("skills.pending_changed", payload) + except Exception: + logger.debug("auto-applied-update notification failed", exc_info=True) + + if _gw_loop is not None and not _gw_loop.is_closed(): + try: + _gw_loop.call_soon_threadsafe(_emit) + except RuntimeError: # pragma: no cover - loop closing + pass + else: + _emit() + except Exception: + logger.debug("auto-applied-update notification failed", exc_info=True) + + set_update_auto_applied_hook(_on_update_auto_applied) except Exception: logger.debug("Could not register pending-skill staged hook", exc_info=True) diff --git a/src/kiro_crew/history.py b/src/kiro_crew/history.py index f0542877365..0fac3c065f0 100644 --- a/src/kiro_crew/history.py +++ b/src/kiro_crew/history.py @@ -15,6 +15,7 @@ import math import os import re +import secrets import threading import time as _time from collections import OrderedDict @@ -5273,6 +5274,8 @@ def _stage_skill_update( triggers: str, procedure_md: str, scripts: "list[dict] | None" = None, + scripts_supplied: bool = False, + allow_auto_apply: bool = True, ) -> None: """Stage a pending UPDATE candidate for an existing auto-skill. @@ -5281,7 +5284,25 @@ def _stage_skill_update( 90s, fail-open); (c) use the redacted merge as the proposed body, else fall back to the candidate's own procedure (also on oversize); (d) stage under ``-update`` with ``kind='update'`` metadata; (e) SEL - audit with outcome ``staged_update``.""" + audit with outcome ``staged_update``. + + When approval is disabled AND the candidate is prose-only (it supplied + no scripts at all — ``scripts_supplied`` mirrors the NEW-candidate rule + that a candidate whose scripts were ALL rejected by the validator still + never auto-publishes), the staged candidate is immediately promoted via + ``auto_apply_pending_update`` (same guards + version snapshot as a human + approval) and audited as ``auto_applied_update``. If that promotion + fails for any reason the candidate stays staged for normal review. + + ``allow_auto_apply=False`` forces the staging path even when the + conditions above would promote: the caller passes it whenever + auto-refine is enabled, because the refine path writes through + ``update_auto_skill`` — unlocked, no version bump, no snapshot — and a + refine of the target can arrive from any concurrent session. A refine + does not advance the version frontmatter, so even a promotion + serialized behind it passes the stale-base check and the two writers + last-write-win. Left staged, the candidate is reviewed against + whatever body the refine left behind instead.""" loader = self._skills_loader if loader is None: return @@ -5382,6 +5403,20 @@ def _redact(text: object) -> str: _live_description = _frontmatter_value(live_body, "description") _staged_triggers = _merge_trigger_lists(_live_triggers, triggers) _staged_description = description or _live_description + # Prose-only + approval disabled → promote immediately after staging — + # unless the caller vetoed it (a same-result refine of this target + # would overwrite the promotion without a snapshot; see docstring). + auto_apply = ( + allow_auto_apply + and not self._approval_required + and not scripts + and not scripts_supplied + ) + # Collision-proof ownership token: only the .meta.json THIS call writes + # carries it, so the promotion step below can prove the pending entry is + # the one just staged (a deferred re-stage writes nothing, and an + # unrelated candidate staged the same second shares the timestamp). + stage_token = secrets.token_hex(16) if auto_apply else None name = loader.stage_skill_candidate( _update_slug, description=_staged_description, @@ -5392,6 +5427,12 @@ def _redact(text: object) -> str: kind="update", target=target_key, base_version=base_version, + # Suppress the awaiting-review notification when this flow is about + # to promote the candidate itself; if the promotion fails, the + # review request is re-fired below so the candidate never sits + # invisible in the queue. + notify=not auto_apply, + stage_token=stage_token, ) if name: logger.info("Staged skill update %s (target %s) from session %s", name, target_key, key) @@ -5407,6 +5448,13 @@ def _redact(text: object) -> str: "merged": used_merge, }, ) + if auto_apply: + self._auto_apply_staged_update( + key=key, + staged_name=name, + target_key=target_key, + stage_token=stage_token, + ) else: logger.info("Skill update staging rejected for target '%s'", target_key) sel().log_tool_invocation( @@ -5417,6 +5465,128 @@ def _redact(text: object) -> str: metadata={"slug": _update_slug, "reason": "creation_failed"}, ) + def _auto_apply_staged_update( + self, + *, + key: str, + staged_name: str, + target_key: str, + stage_token: str | None, + ) -> None: + """Promote a just-staged prose-only update immediately (approval off). + + Delegates to ``SkillsLoader.auto_apply_pending_update`` so every guard + of a human approval applies (staleness, layout, redaction, version + snapshot). Fails SAFE: any refusal or error leaves the candidate staged + and re-fires the awaiting-review notification that staging suppressed. + """ + loader = self._skills_loader + if loader is None or not stage_token: + return + staged_slug = staged_name.split("/", 1)[-1] + # Identity check: when the pending queue already holds a candidate for + # every slug in the collision family, ``stage_skill_candidate`` returns + # the name WITHOUT staging (deferred re-stage). Promoting that slug + # would apply a DIFFERENT, previously-staged candidate — so only + # proceed when the pending entry carries the random ``stage_token`` + # written by the staging call this flow just made. The token is only + # ever written when staging actually wrote the candidate, so it cannot + # match a deferred re-stage or an unrelated same-second candidate. + is_ours = False + lookup_failed = False + try: + pend = loader.get_pending_skill(staged_slug) + is_ours = ( + pend is not None + and pend.get("kind") == "update" + and pend.get("meta", {}).get("stage_token") == stage_token + ) + except Exception: + # Transient lookup failure — the candidate this flow just staged + # (with its notification suppressed) is very likely still pending, + # so it must not be left invisible. Distinguish from a genuine + # ownership mismatch, where the pre-existing candidate already + # fired its own notification at staging time. + is_ours = False + lookup_failed = True + if not is_ours: + # A declined promotion is a promotion decision too — audit it so + # the trail explains why the staged candidate was left pending. + sel().log_tool_invocation( + session_key=key, + tool_name="auto_skill_create", + tool_kind="skills", + outcome="auto_apply_failed", + metadata={ + "name": staged_name, + "target": target_key, + "reason": "pending_lookup_failed" if lookup_failed else "ownership_mismatch", + }, + ) + if lookup_failed: + # Re-fire the awaiting-review notification staging suppressed: + # the candidate stays pending and would otherwise sit invisible + # in the queue. emit_pending_staged no-ops if the slug turns + # out not to be pending after all. + try: + loader.emit_pending_staged(staged_slug) + except Exception: + logger.debug("Pending-staged re-notify failed", exc_info=True) + return + applied: "tuple[str, int] | None" = None + try: + applied = loader.auto_apply_pending_update( + staged_slug, expected_stage_token=stage_token + ) + except Exception: + logger.warning( + "Auto-apply failed for staged update %s; leaving it pending review", + staged_name, + exc_info=True, + ) + applied = None + if applied: + applied_name, new_version = applied + logger.info( + "Auto-applied skill update %s -> v%d (approval disabled)", + applied_name, + new_version, + ) + sel().log_tool_invocation( + session_key=key, + tool_name="auto_skill_create", + tool_kind="skills", + outcome="auto_applied_update", + metadata={ + "name": staged_name, + "target": target_key, + "new_version": new_version, + }, + ) + else: + # Audit the failed promotion decision: the success branch records + # ``auto_applied_update``, so a refusal/exception must leave a SEL + # trace too — otherwise the decision is invisible to `security + # audit` and the promotion appears never to have been attempted. + sel().log_tool_invocation( + session_key=key, + tool_name="auto_skill_create", + tool_kind="skills", + outcome="auto_apply_failed", + metadata={ + "name": staged_name, + "target": target_key, + "reason": "left_pending_for_review", + }, + ) + # The candidate is still pending, so surface the review request + # that staging suppressed — otherwise it sits invisible in the + # queue on an instance whose user never opens it. + try: + loader.emit_pending_staged(staged_slug) + except Exception: + logger.debug("Pending-staged re-notify failed", exc_info=True) + def _process_auto_skills(self, result: dict, key: str) -> None: """Extract + write auto-generated skills from the consolidation result. @@ -5532,6 +5702,17 @@ def _redact(text: object) -> str: elif verdict == VERDICT_UPDATE and target: # Same skill, new requirements worth folding in — stage a # pending UPDATE candidate rather than dropping the learning. + # Auto-apply is disabled outright whenever auto-refine is + # enabled: the refine path overwrites live through + # ``update_auto_skill`` — unlocked, no version bump, no + # snapshot — and a refine of this target can arrive from ANY + # concurrent session, not just this result (a result-scoped + # veto cannot see those). Because a refine leaves the version + # frontmatter unchanged, even a serialized promotion passes + # the stale-base check and the two writers last-write-win, + # silently discarding one side. With refine on, the update + # therefore always stages for human review, where it is + # applied against whatever body the refine left behind. self._stage_skill_update( key=key, target_key=target, @@ -5539,6 +5720,8 @@ def _redact(text: object) -> str: triggers=triggers, procedure_md=procedure_md, scripts=valid_scripts or None, + scripts_supplied=scripts_supplied, + allow_auto_apply=not self._auto_refine_enabled, ) else: provenance = AutoSkillProvenance( diff --git a/src/kiro_crew/skills.py b/src/kiro_crew/skills.py index fd1e619aa36..5abc3bb87d2 100644 --- a/src/kiro_crew/skills.py +++ b/src/kiro_crew/skills.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import contextlib import difflib import fnmatch import hashlib @@ -19,6 +20,7 @@ from pathlib import Path from typing import Callable, Iterator +from kiro_crew import platform_compat from kiro_crew.atomic_write import atomic_write from kiro_crew.config.loader import KiroCrewConfig, config_dir from kiro_crew.cron import referenced_skill_names @@ -141,6 +143,22 @@ def _matches_any(path: str, globs: list[str]) -> bool: # Cap on retained per-skill version snapshots; oldest are pruned past this. MAX_SKILL_VERSIONS = 20 +# Per-target promotion lock files. A dot-prefixed dir so it is pruned from +# skill discovery. Layout: ``auto/.locks/.lock`` — one advisory +# lock file per live auto-skill, held for the duration of a candidate +# promotion (``approve_pending_update`` / ``auto_apply_pending_update``) so +# two processes promoting to the SAME target serialize instead of both +# passing the version check and last-write-winning. +AUTO_LOCKS_DIRNAME = ".locks" + +# Bounded promotion-lock acquisition: poll ``LOCK_EX | LOCK_NB`` rather than +# blocking, so a hung (alive but stuck) holder cannot stall promotion forever; +# on timeout the promotion is REFUSED and the candidate stays pending +# (fail-safe). A crashed holder is not a concern: ``flock`` is released by the +# kernel when the holding process dies, so the next acquire succeeds. +_PROMOTE_LOCK_TIMEOUT_S = 10.0 +_PROMOTE_LOCK_POLL_S = 0.05 + # ── Pending-staged observer hook ────────────────────────────────────────────── # A candidate can be staged by ANY ``SkillsLoader`` instance (consolidation uses # the ContextBuilder's loader; dashboard requests build their own), so the @@ -210,6 +228,38 @@ def _emit_pending_consumed(payload: dict) -> None: logger.debug("pending-consumed hook failed", exc_info=True) +# Observer for prose-only updates promoted WITHOUT review (approval disabled). +# Separate from the pending-staged hook because the message class differs: this +# is informational ("skill X auto-updated to vN", prior version restorable), not +# a review request — reusing the staged hook would show "awaiting review" for a +# candidate that already went live. Module-level for the same reason as above. +_UPDATE_AUTO_APPLIED_HOOK: "Callable[[dict], None] | None" = None + + +def set_update_auto_applied_hook(fn: "Callable[[dict], None] | None") -> None: + """Register (or clear, with ``None``) the update-auto-applied observer. + + Called once at gateway boot, next to ``set_pending_staged_hook``. Idempotent. + """ + global _UPDATE_AUTO_APPLIED_HOOK + _UPDATE_AUTO_APPLIED_HOOK = fn + + +def _emit_update_auto_applied(payload: dict) -> None: + """Invoke the update-auto-applied hook, swallowing every failure. + + The update is already live by the time this runs; a broken observer must + never turn a successful promotion into a failure. + """ + fn = _UPDATE_AUTO_APPLIED_HOOK + if fn is None: + return + try: + fn(payload) + except Exception: # pragma: no cover - defensive + logger.debug("update-auto-applied hook failed", exc_info=True) + + # Derived lifecycle states for auto-skills (not persisted — computed from # usage recency at lifecycle-run time). SKILL_STATE_ACTIVE = "active" @@ -2325,6 +2375,64 @@ def run_skill_lifecycle( def _pending_root(self) -> Path: return self._dir / AUTO_SKILL_NAMESPACE / AUTO_PENDING_DIRNAME + def _locks_root(self) -> Path: + return self._dir / AUTO_SKILL_NAMESPACE / AUTO_LOCKS_DIRNAME + + @contextlib.contextmanager + def _promotion_lock(self, target_slug: str) -> Iterator[bool]: + """Per-target cross-process promotion lock; yields True when held. + + Serializes candidate promotion onto one live auto-skill across + processes: an advisory ``flock(LOCK_EX)`` on + ``auto/.locks/.lock``. Two processes approving/auto- + applying updates to the SAME target from the same base would both pass + the version (stale-base) check and last-write-win; under the lock the + second promoter re-reads the live version after the first commits and + is refused as stale. + + Acquisition is a bounded non-blocking poll (``LOCK_NB`` up to + ``_PROMOTE_LOCK_TIMEOUT_S``), never a blocking wait, so a hung holder + cannot stall promotion indefinitely — on timeout the context yields + False and the caller refuses the promotion, leaving the candidate + pending (fail-safe). A CRASHED holder cannot deadlock promotion at + all: POSIX ``flock`` locks belong to the open file description and are + released by the kernel when the holding process dies, and Windows + ``msvcrt`` region locks are likewise released on process exit. Both + paths go through :func:`platform_compat.try_acquire_lock`, so the lock + is effective cross-platform (not a Windows no-op). + + The lock file itself is never deleted (unlinking a lock file while a + waiter holds its fd open reintroduces the race the lock closes). + """ + lock_path = self._locks_root() / f"{target_slug}.lock" + try: + lock_path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o600) + except OSError: + logger.warning( + "Could not open promotion lock file for %r; refusing promotion", target_slug + ) + yield False + return + acquired = False + try: + deadline = time.monotonic() + _PROMOTE_LOCK_TIMEOUT_S + while True: + if platform_compat.try_acquire_lock(fd, exclusive=True): + acquired = True + break + if time.monotonic() >= deadline: + break + time.sleep(_PROMOTE_LOCK_POLL_S) + yield acquired + finally: + if acquired: + platform_compat.release_lock(fd) + try: + os.close(fd) + except OSError: + pass + def stage_skill_candidate( self, slug: str, @@ -2338,6 +2446,8 @@ def stage_skill_candidate( kind: str = "new", target: str | None = None, base_version: int | None = None, + notify: bool = True, + stage_token: str | None = None, ) -> str | None: """Write a skill candidate to the pending queue (not live). @@ -2354,6 +2464,12 @@ def stage_skill_candidate( version the merge was based on. These are written into ``.meta.json`` (``kind`` always; ``target`` / ``base_version`` only when provided) so existing new-candidate callers are unaffected. + + ``notify=False`` suppresses the awaiting-review observer notification. + Pass it ONLY when the caller intends to promote the candidate in the + same flow (see ``auto_apply_pending_update``); if that promotion fails, + the caller must fire ``emit_pending_staged`` so the still-pending + candidate does not sit invisible in the queue. """ if not _AUTO_NAME_PATTERN.match(slug): logger.warning("Rejected pending skill: slug %r failed validation", slug) @@ -2432,6 +2548,14 @@ def stage_skill_candidate( meta["target"] = target if base_version is not None: meta["base_version"] = base_version + if stage_token is not None: + # Collision-proof ownership token: written ONLY when this call + # actually wrote the candidate to disk. The deferred re-stage + # path (collision family exhausted) returns a name WITHOUT + # writing, so a caller that intends to promote what it just + # staged must verify this token — timestamps cannot distinguish + # an unrelated candidate staged in the same second. + meta["stage_token"] = stage_token (pdir / ".meta.json").write_text(json.dumps(meta, indent=2), encoding="utf-8") except Exception: # A partial write (e.g. disk full) must not leave a CLAIMED but empty @@ -2447,8 +2571,10 @@ def stage_skill_candidate( # notification + a ``skills.pending_changed`` WS event) so a candidate # awaiting review surfaces instead of sitting unseen in the queue. Fired # for BOTH new and update candidates, from every producer that stages - # through this choke point. Best-effort: an observer failure must never - # fail the staging that already succeeded on disk. + # through this choke point (unless the caller suppressed it with + # ``notify=False`` because it is about to promote the candidate itself). + # Best-effort: an observer failure must never fail the staging that + # already succeeded on disk. # # ``description``/``triggers`` ride along because the observer's only # other option is to re-read ``.meta.json`` off disk (a second read of @@ -2456,19 +2582,47 @@ def stage_skill_candidate( # notification can only say THAT a skill was generated, never what it # does, which is the one fact a reviewer needs to decide whether to open # the queue at all. + if notify: + _emit_pending_staged( + { + "name": name, + "slug": slug, + "kind": kind or "new", + "target": target, + "source": source, + "has_scripts": bool(script_names), + "description": description, + "triggers": triggers, + } + ) + return name + + def emit_pending_staged(self, slug: str) -> None: + """Fire the awaiting-review notification for an already-staged candidate. + + For callers that staged with ``notify=False`` intending an immediate + promotion that then FAILED: the candidate is still pending, so the + review request it would normally have raised is fired now (from the + on-disk ``.meta.json``, the same fields the staging-time payload + carries). No-op if the slug is unsafe or not pending. + """ + if not self._is_pending_slug_safe(slug): + return + if not (self._pending_root() / slug / "SKILL.md").exists(): + return + meta = self._read_pending_meta(slug) _emit_pending_staged( { - "name": name, + "name": meta.get("name", f"{AUTO_SKILL_NAMESPACE}/{slug}"), "slug": slug, - "kind": kind or "new", - "target": target, - "source": source, - "has_scripts": bool(script_names), - "description": description, - "triggers": triggers, + "kind": meta.get("kind", "new"), + "target": meta.get("target"), + "source": meta.get("source", ""), + "has_scripts": bool(meta.get("has_scripts")), + "description": meta.get("description", ""), + "triggers": meta.get("triggers", ""), } ) - return name def _read_pending_meta(self, slug: str) -> dict: mf = self._pending_root() / slug / ".meta.json" @@ -2980,9 +3134,64 @@ def _resolve_snapshot_version(self, versions_dir: Path, fm_version: int) -> int: ) return highest + 1 - def approve_pending_update(self, slug: str) -> str | None: + def approve_pending_update( + self, slug: str, *, refuse_scripts: bool = False, expected_stage_token: str | None = None + ) -> str | None: """Promote a pending UPDATE candidate over its live target auto-skill. + Thin locking wrapper: derives the promotion target from the + candidate's ``.meta.json`` and runs the entire promotion + (:meth:`_approve_pending_update_locked`) under the per-target + cross-process :meth:`_promotion_lock`, so concurrent promotions onto + the same live skill serialize — the version (stale-base) check, the + ``refuse_scripts`` re-check, the live overwrite/copy, and the + candidate delete all execute under one holder. Failing to acquire the + lock refuses the promotion (returns ``None``; the candidate stays + pending and reviewable). + + ``expected_stage_token``: when set, the candidate's identity is + re-verified INSIDE the lock — the pending ``.meta.json`` must carry + this exact token or the promotion is refused. Closes the + check-then-promote window where a concurrent dismiss + same-slug + re-stage swaps in a different (unreviewed) candidate between the + caller's pre-lock ownership check and the locked promotion. + """ + if not self._is_pending_slug_safe(slug): + return None + meta = self._read_pending_meta(slug) + target = meta.get("target") + if not isinstance(target, str) or not target: + return None + target_slug = self._auto_slug_from_name(target) + if not self._is_pending_slug_safe(target_slug): + return None + with self._promotion_lock(target_slug) as acquired: + if not acquired: + logger.warning( + "Refusing to approve update %s: promotion lock for %r not " + "acquired within %.0fs (concurrent promotion in flight?)", + slug, + target_slug, + _PROMOTE_LOCK_TIMEOUT_S, + ) + return None + return self._approve_pending_update_locked( + slug, + refuse_scripts=refuse_scripts, + expected_stage_token=expected_stage_token, + ) + + def _approve_pending_update_locked( + self, slug: str, *, refuse_scripts: bool = False, expected_stage_token: str | None = None + ) -> str | None: + """Locked body of :meth:`approve_pending_update`. + + MUST be called with the per-target :meth:`_promotion_lock` held: the + stale-base version check, the ``refuse_scripts`` re-check, and the + live writes below are only race-free while this process is the sole + promoter for the target. All candidate state is (re-)read here, under + the lock — nothing is trusted from before acquisition. + Preconditions (all checked BEFORE any live mutation; a failure here leaves BOTH the live skill and the candidate untouched, returns None): the slug is safe, the candidate has a ``SKILL.md``, its ``.meta.json`` @@ -3007,9 +3216,32 @@ def approve_pending_update(self, slug: str) -> str | None: meta = self._read_pending_meta(slug) if meta.get("kind") != "update": return None + if expected_stage_token is not None and meta.get("stage_token") != expected_stage_token: + # Identity re-check under the lock: the candidate occupying this + # slug is NOT the one the caller staged (concurrent dismiss + + # same-slug re-stage). Promoting it would ship an unreviewed + # replacement — refuse and leave it pending for normal review. + logger.warning( + "Refusing unattended promotion of %s: pending candidate's " + "stage token does not match the staging flow's token " + "(candidate swapped since the pre-lock check)", + slug, + ) + return None target = meta.get("target") if not isinstance(target, str) or not target: return None + if refuse_scripts and (meta.get("has_scripts") or (src / "scripts").exists()): + # Unattended promotion (approval disabled) must never ship scripts: + # they always require human review. This early check handles the + # common case cleanly (no mutation yet); the copy step below + # re-enforces it atomically for scripts injected mid-promotion. + logger.info( + "Refusing to auto-apply update %s: candidate bundles scripts " + "(scripts always require review)", + slug, + ) + return None target_slug = self._auto_slug_from_name(target) if not self._is_pending_slug_safe(target_slug): return None @@ -3160,7 +3392,39 @@ def _restore_redacted() -> None: # and then failing on a later file would roll SKILL.md back while leaving # the replacement script live — an internally inconsistent skill. overwritten: dict[Path, tuple[bytes, int]] = {} - if src_scripts.is_dir(): + if refuse_scripts and src_scripts.is_dir(): + # TOCTOU closure: scripts appeared AFTER the entry precondition (a + # concurrent writer added them mid-promotion). Unattended promotion + # must never ship a script no human reviewed — abort and roll back + # the live SKILL.md written in (f), leaving the candidate pending. + try: + atomic_write(live_skill, live_prev) + except OSError: + logger.error( + "Update %s aborted on late scripts AND the live SKILL.md " + "could not be restored; the snapshot remains at %s", + target_name, + snapshot, + ) + else: + try: + snapshot.unlink() + except OSError: + pass + _restore_redacted() + logger.warning( + "Refusing to auto-apply update %s: scripts appeared during " + "promotion (scripts always require review)", + target_name, + ) + return None + # Structural closure of the remaining check-to-copy window: under + # ``refuse_scripts`` the copy path below is UNREACHABLE, so a script + # dir injected after the re-check above can never go live — at worst + # it is discarded with the candidate delete, never executed. The + # promotion lock serializes concurrent promoters; this guard also + # covers non-promoter writers (a staging process) racing the copy. + if not refuse_scripts and src_scripts.is_dir(): live_scripts = live_dir / "scripts" try: live_scripts.mkdir(parents=True, exist_ok=True) @@ -3262,6 +3526,52 @@ def _restore_redacted() -> None: ) return target_name + def auto_apply_pending_update( + self, slug: str, *, expected_stage_token: str | None = None + ) -> "tuple[str, int] | None": + """Promote a staged prose-only UPDATE candidate without human review. + + Used when ``skills.approval_required`` is off: prose-only updates then + go live the same way prose-only NEW candidates do, instead of rotting + in the pending queue on an instance that opted out of review. Every + write and guard is delegated to ``approve_pending_update`` (staleness / + base_version, symlink + layout guards, redaction, version snapshot, + pruning, SEL audit) — this wrapper only adds the informational + notification; the script refusal is enforced INSIDE + ``approve_pending_update`` via ``refuse_scripts=True`` so it holds + atomically through the promotion (a concurrent writer adding scripts + mid-flight aborts the promotion instead of shipping them live). + + Script-bearing candidates are REFUSED (returns ``None``, candidate + stays pending): scripts always require human review regardless of the + approval flag, mirroring the NEW-candidate rule. + + On success emits the update-auto-applied observer notification (an + informational "went live" signal, NOT a review request) and returns + ``(target_name, new_live_version)``. Any failure returns ``None`` and + leaves the candidate staged for normal review. + """ + if not self._is_pending_slug_safe(slug): + return None + # Read the metadata BEFORE approval deletes the pending dir. + meta = self._read_pending_meta(slug) + name = self.approve_pending_update( + slug, refuse_scripts=True, expected_stage_token=expected_stage_token + ) + if not name: + return None + new_version = self.get_auto_skill_version(name) + _emit_update_auto_applied( + { + "name": name, + "slug": slug, + "target": name, + "new_version": new_version, + "description": meta.get("description", ""), + } + ) + return name, new_version + def approve_pending_skill(self, slug: str) -> str | None: """Promote a pending candidate to a live auto-skill. diff --git a/test/test_skill_update_auto_apply.py b/test/test_skill_update_auto_apply.py new file mode 100644 index 00000000000..903f9f92a8d --- /dev/null +++ b/test/test_skill_update_auto_apply.py @@ -0,0 +1,863 @@ +"""Tests: auto-apply of prose-only skill UPDATE candidates when approval is off. + +When ``skills.approval_required`` is false, prose-only NEW candidates already +go live without review — these tests cover the matching UPDATE behavior: a +prose-only update candidate is staged and immediately promoted through +``approve_pending_update`` (same guards, version snapshot, pruning), audited as +``auto_applied_update`` and announced via the informational auto-applied hook +instead of a review request. Script-bearing updates and approval-on instances +keep the staging behavior, and any promotion failure fails SAFE (candidate +stays pending, review notification fires). +""" + +from __future__ import annotations + +import os +import threading +import time +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from kiro_crew import flock_compat +from kiro_crew import skills as S +from kiro_crew.history import VERDICT_UPDATE, HistoryConsolidator +from kiro_crew.skills import AutoSkillProvenance, SkillsLoader + +_NEW_STEPS = "## Steps\n\n1. the new way\n" + + +@pytest.fixture() +def loader(tmp_path): + return SkillsLoader(skills_path=tmp_path / "skills", install_builtins=False) + + +@pytest.fixture(autouse=True) +def _clear_hooks(): + """Never leak a hook across tests (module-level global state).""" + S.set_pending_staged_hook(None) + S.set_update_auto_applied_hook(None) + yield + S.set_pending_staged_hook(None) + S.set_update_auto_applied_hook(None) + + +def _write_live(loader, slug, *, version=1, body="original body"): + """Write a live auto-skill directly (frontmatter version included).""" + live = loader._dir / "auto" / slug + live.mkdir(parents=True, exist_ok=True) + content = ( + "---\n" + f"name: auto/{slug}\n" + "description: live desc\n" + "triggers: t\n" + "source: auto\n" + "created_at: 2020-01-01T00:00:00+00:00\n" + f"version: {version}\n" + "---\n\n" + f"# {slug}\n\n{body}\n" + ) + (live / "SKILL.md").write_text(content, encoding="utf-8") + loader._invalidate_iter_cache() + return live + + +def _prov() -> AutoSkillProvenance: + return AutoSkillProvenance( + session_key="s", + created_at=datetime.now(tz=timezone.utc).isoformat(timespec="seconds"), + ) + + +def _mk(loader, *, approval_required, auto_refine_enabled=False): + c = HistoryConsolidator( + log=MagicMock(), + memory=MagicMock(), + skills_loader=loader, + auto_skills_enabled=True, + approval_required=approval_required, + auto_refine_enabled=auto_refine_enabled, + ) + c._event_loop = None # skip the LLM merge; candidate body is used as-is + return c + + +def _sel_recorder(recorded): + ctx = patch("kiro_crew.history.sel") + mock = ctx.start() + mock.return_value.log_tool_invocation = lambda **k: recorded.append(k) + return ctx + + +def _stage_update(c, *, scripts=None, scripts_supplied=False): + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="new desc", + triggers="new trigger", + procedure_md=_NEW_STEPS, + scripts=scripts, + scripts_supplied=scripts_supplied, + ) + + +def _pending_slugs(loader): + return [p["slug"] for p in loader.list_pending_skills()] + + +# ── (a) prose-only update auto-applies when approval is off ── + + +def test_prose_only_update_auto_applies_when_approval_off(loader): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=False) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + _stage_update(c) + finally: + ctx.stop() + + # Live skill carries the update and its version incremented. + body = loader.read_auto_skill_body("auto/deploy-helper") + assert body is not None and "the new way" in body + assert loader.get_auto_skill_version("auto/deploy-helper") == 2 + # The prior version was snapshotted for rollback. + snapshot = loader._dir / "auto" / "deploy-helper" / ".versions" / "v1-SKILL.md" + assert snapshot.exists() + assert "original body" in snapshot.read_text(encoding="utf-8") + # The candidate did not stay in the queue. + assert _pending_slugs(loader) == [] + # Audited as an auto-applied update. + applied = [r for r in recorded if r.get("outcome") == "auto_applied_update"] + assert applied and applied[0]["metadata"]["new_version"] == 2 + assert applied[0]["metadata"]["target"] == "auto/deploy-helper" + # Informational notification fired; the review request did NOT. + assert len(applied_seen) == 1 + assert applied_seen[0]["target"] == "auto/deploy-helper" + assert applied_seen[0]["new_version"] == 2 + assert staged_seen == [] + + +# ── (b) script-bearing update still stages when approval is off ── + + +def test_script_bearing_update_still_stages_when_approval_off(loader): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=False) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + _stage_update( + c, + scripts=[{"filename": "go.py", "content": "print('hi')\n"}], + scripts_supplied=True, + ) + finally: + ctx.stop() + + # Candidate is queued for review; live is untouched. + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + body = loader.read_auto_skill_body("auto/deploy-helper") + assert body is not None and "original body" in body + # Review request fired; no auto-apply happened. + assert len(staged_seen) == 1 and staged_seen[0]["has_scripts"] is True + assert applied_seen == [] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + + +# ── (c) prose update still stages when approval is on ── + + +def test_prose_update_still_stages_when_approval_on(loader): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=True) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + _stage_update(c) + finally: + ctx.stop() + + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + assert len(staged_seen) == 1 + assert applied_seen == [] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + + +# ── (c2) same-result refine of the target vetoes auto-apply ── + + +def test_update_stays_staged_when_same_result_refines_target(loader, monkeypatch): + """A ``new_skill`` deduped as an UPDATE of a target that the same result + ALSO refines must not auto-apply: the refine path overwrites live through + ``update_auto_skill`` — no version snapshot, body derived from the + pre-update skill — so an immediate promotion would be silently destroyed. + The update stays staged for review against the refined skill.""" + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=False, auto_refine_enabled=True) + monkeypatch.setattr( + c, "_dedupe_candidate", lambda s, d, t: (VERDICT_UPDATE, "auto/deploy-helper") + ) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._process_auto_skills( + { + "new_skill": { + "slug": "deploy-helper-2", + "description": "new desc", + "triggers": "new trigger", + "procedure_md": _NEW_STEPS, + }, + "refined_skill": { + "name": "auto/deploy-helper", + "description": "refined desc", + "triggers": "t", + "procedure_md": "## Steps\n\n1. the refined way\n", + }, + }, + "sess", + ) + finally: + ctx.stop() + + # The update candidate stayed in the queue and its body survives for review. + assert _pending_slugs(loader) == ["deploy-helper-update"] + pend = loader.get_pending_skill("deploy-helper-update") + assert pend is not None and "the new way" in pend["content"] + # No promotion happened; the refine result owns the live body. + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + assert applied_seen == [] + body = loader.read_auto_skill_body("auto/deploy-helper") + assert body is not None and "the refined way" in body + assert "the new way" not in body + # The review request fired (staging did not suppress it). + assert len(staged_seen) == 1 + assert staged_seen[0]["slug"] == "deploy-helper-update" + + +def test_refine_enabled_stages_update_even_for_other_target(loader, monkeypatch): + """The guard is config-scoped, not result-scoped: with auto-refine enabled, + a refine of THIS update's target can arrive from any concurrent session — + invisible to this result — so auto-apply is disabled even when this + result's own ``refined_skill`` names a different skill. The update stages + for review; the unrelated refine still lands on its own target.""" + _write_live(loader, "deploy-helper", version=1) + _write_live(loader, "other-skill", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=False, auto_refine_enabled=True) + monkeypatch.setattr( + c, "_dedupe_candidate", lambda s, d, t: (VERDICT_UPDATE, "auto/deploy-helper") + ) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._process_auto_skills( + { + "new_skill": { + "slug": "deploy-helper-2", + "description": "new desc", + "triggers": "new trigger", + "procedure_md": _NEW_STEPS, + }, + "refined_skill": { + "name": "auto/other-skill", + "description": "refined desc", + "triggers": "t", + "procedure_md": "## Steps\n\n1. refined other\n", + }, + }, + "sess", + ) + finally: + ctx.stop() + + # No unattended promotion while refine is enabled; the update is staged. + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + assert applied_seen == [] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + # The unrelated refine landed on its own target. + other = loader.read_auto_skill_body("auto/other-skill") + assert other is not None and "refined other" in other + # The review request fired for the staged update. + assert len(staged_seen) == 1 + assert staged_seen[0]["slug"] == "deploy-helper-update" + + +def test_concurrent_session_refine_does_not_lose_either_write(loader, monkeypatch): + """Regression (GPT round 9): with auto-refine enabled, a DIFFERENT + session's refine of the same target races the unattended promotion. + ``update_auto_skill`` is unlocked and leaves the version frontmatter + unchanged, so the promotion's stale-base check passes and last-write-wins + silently discards one side. With auto-apply disabled under refine, the + concurrent refine's write survives live and this session's update stays + staged — neither write is lost.""" + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=False, auto_refine_enabled=True) + monkeypatch.setattr( + c, "_dedupe_candidate", lambda s, d, t: (VERDICT_UPDATE, "auto/deploy-helper") + ) + + # Interleave the other session's refine right after this session stages — + # inside the window where the pre-guard code promoted next. On the old + # code the promotion then overwrote live (base_version 1 == live version + # 1, staleness check passes) and the refine's write was discarded. + real_stage = loader.stage_skill_candidate + + def _stage_then_concurrent_refine(*args, **kwargs): + name = real_stage(*args, **kwargs) + assert loader.update_auto_skill( + "auto/deploy-helper", + description="refined by other session", + triggers="t", + procedure_md="## Steps\n\n1. the other session way\n", + provenance=_prov(), + ) + return name + + monkeypatch.setattr(loader, "stage_skill_candidate", _stage_then_concurrent_refine) + + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._process_auto_skills( + { + "new_skill": { + "slug": "deploy-helper-2", + "description": "new desc", + "triggers": "new trigger", + "procedure_md": _NEW_STEPS, + } + }, + "sess", + ) + finally: + ctx.stop() + + # No unattended promotion raced the refine. + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + assert applied_seen == [] + # The concurrent session's refine survives live... + body = loader.read_auto_skill_body("auto/deploy-helper") + assert body is not None and "the other session way" in body + assert "the new way" not in body + # ...and this session's update was not lost: it is staged for review. + assert _pending_slugs(loader) == ["deploy-helper-update"] + pend = loader.get_pending_skill("deploy-helper-update") + assert pend is not None and "the new way" in pend["content"] + assert len(staged_seen) == 1 + + +# ── (d) auto-approve failure leaves the candidate staged ── + + +def test_auto_apply_failure_leaves_candidate_staged(loader, monkeypatch): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + applied_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + S.set_update_auto_applied_hook(applied_seen.append) + # Promotion refuses (any approve_pending_update guard tripping). + monkeypatch.setattr(loader, "approve_pending_update", lambda slug: None) + + c = _mk(loader, approval_required=False) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + _stage_update(c) + finally: + ctx.stop() + + # Fail SAFE: the candidate stays pending, live untouched. + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + # The review request suppressed at staging time was re-fired, so the + # still-pending candidate does not sit invisible in the queue. + assert len(staged_seen) == 1 + assert staged_seen[0]["slug"] == "deploy-helper-update" + assert staged_seen[0]["kind"] == "update" + assert applied_seen == [] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + # The staging itself is still audited. + assert [r for r in recorded if r.get("outcome") == "staged_update"] + + +def test_auto_apply_exception_leaves_candidate_staged(loader, monkeypatch): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + + def boom(slug): + raise RuntimeError("disk went away") + + monkeypatch.setattr(loader, "auto_apply_pending_update", boom) + + c = _mk(loader, approval_required=False) + ctx = _sel_recorder([]) + try: + _stage_update(c) + finally: + ctx.stop() + + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + assert len(staged_seen) == 1 + + +# ── (e) a candidate that SUPPLIED scripts never auto-applies, even when the +# validator rejected every one of them ── + + +def test_rejected_scripts_candidate_still_stages_when_approval_off(loader): + _write_live(loader, "deploy-helper", version=1) + applied_seen: list[dict] = [] + S.set_update_auto_applied_hook(applied_seen.append) + + c = _mk(loader, approval_required=False) + ctx = _sel_recorder([]) + try: + # scripts=None (all rejected by the validator) but scripts_supplied=True: + # the candidate wanted scripts, so it must never auto-publish. + _stage_update(c, scripts=None, scripts_supplied=True) + finally: + ctx.stop() + + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + assert applied_seen == [] + + +# ── loader-level: auto_apply_pending_update guards ── + + +def _stage_candidate(loader, *, scripts=None, base_version=1, notify=True): + return loader.stage_skill_candidate( + "deploy-helper-update", + description="updated", + triggers="t", + procedure_md=_NEW_STEPS, + provenance=_prov(), + scripts=scripts, + kind="update", + target="auto/deploy-helper", + base_version=base_version, + notify=notify, + ) + + +def test_loader_auto_apply_promotes_and_emits(loader): + _write_live(loader, "deploy-helper", version=1) + applied_seen: list[dict] = [] + S.set_update_auto_applied_hook(applied_seen.append) + assert _stage_candidate(loader, notify=False) == "auto/deploy-helper-update" + + result = loader.auto_apply_pending_update("deploy-helper-update") + + assert result == ("auto/deploy-helper", 2) + assert loader.get_auto_skill_version("auto/deploy-helper") == 2 + assert _pending_slugs(loader) == [] + assert len(applied_seen) == 1 + assert applied_seen[0]["target"] == "auto/deploy-helper" + assert applied_seen[0]["new_version"] == 2 + + +def test_loader_auto_apply_refuses_script_bearing_candidate(loader): + _write_live(loader, "deploy-helper", version=1) + applied_seen: list[dict] = [] + S.set_update_auto_applied_hook(applied_seen.append) + _stage_candidate( + loader, scripts=[{"filename": "go.py", "content": "print('hi')\n"}], notify=False + ) + + assert loader.auto_apply_pending_update("deploy-helper-update") is None + + # Candidate untouched, live untouched, no notification. + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + assert applied_seen == [] + + +def test_loader_auto_apply_refuses_physical_scripts_dir_without_meta(loader): + """Defense in depth: a direct-write candidate whose ``.meta.json`` is + missing or lies about ``has_scripts`` must still be refused when a physical + ``scripts/`` dir exists — otherwise scripts would go live unreviewed.""" + _write_live(loader, "deploy-helper", version=1) + _stage_candidate(loader, notify=False) + pend = loader._pending_root() / "deploy-helper-update" + (pend / "scripts").mkdir() + (pend / "scripts" / "sneaky.py").write_text("print('hi')\n", encoding="utf-8") + + assert loader.auto_apply_pending_update("deploy-helper-update") is None + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + + +def test_loader_auto_apply_toctou_scripts_injected_mid_promotion(loader, monkeypatch): + """TOCTOU closure: scripts injected AFTER the entry precondition but before + the copy step must abort the promotion (live SKILL.md rolled back, snapshot + removed, candidate left pending) — never ship an unreviewed script live.""" + _write_live(loader, "deploy-helper", version=1) + applied_seen: list[dict] = [] + S.set_update_auto_applied_hook(applied_seen.append) + _stage_candidate(loader, notify=False) + pend = loader._pending_root() / "deploy-helper-update" + live_before = ( + loader._dir / "auto" / "deploy-helper" / "SKILL.md" + ).read_text(encoding="utf-8") + + # Simulate the concurrent writer deterministically: inject scripts/ as a + # side effect of the redaction step, which runs after the entry + # precondition and before the copy step. + real_redact = loader._validate_and_redact_candidate + + def _inject_then_redact(src, target_name): + (pend / "scripts").mkdir(exist_ok=True) + (pend / "scripts" / "sneaky.py").write_text("print('hi')\n", encoding="utf-8") + return real_redact(src, target_name) + + monkeypatch.setattr(loader, "_validate_and_redact_candidate", _inject_then_redact) + + assert loader.auto_apply_pending_update("deploy-helper-update") is None + + # Live rolled back byte-identical, no version bump, no snapshot left over. + live_after = ( + loader._dir / "auto" / "deploy-helper" / "SKILL.md" + ).read_text(encoding="utf-8") + assert live_after == live_before + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + versions_dir = loader._dir / "auto" / "deploy-helper" / ".versions" + assert not versions_dir.is_dir() or not any(versions_dir.iterdir()) + # Candidate still pending, injected script NOT live, no notification. + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert not (loader._dir / "auto" / "deploy-helper" / "scripts" / "sneaky.py").exists() + assert applied_seen == [] + + +def test_loader_auto_apply_stale_base_refused_and_stays_pending(loader): + """The staleness guard is inherited from approve_pending_update: a + candidate merged against an older live version is refused, not applied.""" + _write_live(loader, "deploy-helper", version=2) # live moved past base 1 + applied_seen: list[dict] = [] + S.set_update_auto_applied_hook(applied_seen.append) + _stage_candidate(loader, base_version=1, notify=False) + + assert loader.auto_apply_pending_update("deploy-helper-update") is None + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 2 + assert applied_seen == [] + + +# ── notify=False staging + emit_pending_staged re-fire ── + + +def test_stage_notify_false_suppresses_review_notification(loader): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + _stage_candidate(loader, notify=False) + assert staged_seen == [] + + +def test_emit_pending_staged_refires_from_meta(loader): + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + _stage_candidate(loader, notify=False) + + loader.emit_pending_staged("deploy-helper-update") + + assert len(staged_seen) == 1 + payload = staged_seen[0] + assert payload["name"] == "auto/deploy-helper-update" + assert payload["slug"] == "deploy-helper-update" + assert payload["kind"] == "update" + assert payload["target"] == "auto/deploy-helper" + assert payload["has_scripts"] is False + assert payload["description"] == "updated" + + +def test_emit_pending_staged_noop_for_missing_candidate(loader): + staged_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + loader.emit_pending_staged("never-staged") + assert staged_seen == [] + + +# ── per-target promotion lock (cross-process serialization) ── + + +def _stage_named(loader, slug, *, procedure_md=_NEW_STEPS, base_version=1): + return loader.stage_skill_candidate( + slug, + description="updated", + triggers="t", + procedure_md=procedure_md, + provenance=_prov(), + scripts=None, + kind="update", + target="auto/deploy-helper", + base_version=base_version, + notify=False, + ) + + +needs_flock = pytest.mark.skipif( + not flock_compat.HAVE_FCNTL, reason="real flock required (no-op on Windows)" +) + + +@needs_flock +def test_concurrent_same_target_promotions_serialize(loader, monkeypatch): + """Two promoters targeting the same live skill from the same base must + serialize under the per-target lock: the first wins, the second re-reads + the live version under the lock and is refused as stale — never a silent + last-write-wins with both candidates consumed.""" + _write_live(loader, "deploy-helper", version=1) + loader_b = SkillsLoader(skills_path=loader._dir, install_builtins=False) + assert _stage_named(loader, "upd-a") == "auto/upd-a" + assert _stage_named(loader_b, "upd-b", procedure_md="## Steps\n\n1. the B way\n") == ( + "auto/upd-b" + ) + assert sorted(_pending_slugs(loader)) == ["upd-a", "upd-b"] + + a_in_lock = threading.Event() + release_a = threading.Event() + real_redact = loader._validate_and_redact_candidate + + def _hold_inside_lock(src, target_name): + # Runs INSIDE promoter A's critical section: park here so promoter B + # demonstrably contends on the lock while A is mid-promotion. + a_in_lock.set() + assert release_a.wait(timeout=10) + return real_redact(src, target_name) + + monkeypatch.setattr(loader, "_validate_and_redact_candidate", _hold_inside_lock) + + results: dict = {} + t_a = threading.Thread( + target=lambda: results.__setitem__("a", loader.approve_pending_update("upd-a")) + ) + t_b = threading.Thread( + target=lambda: results.__setitem__("b", loader_b.approve_pending_update("upd-b")) + ) + t_a.start() + assert a_in_lock.wait(timeout=10) + t_b.start() + # B is now polling the held lock (poll interval is 0.05s). + time.sleep(0.4) + assert "b" not in results # B must not have promoted while A holds the lock + release_a.set() + t_a.join(timeout=30) + t_b.join(timeout=30) + + assert results["a"] == "auto/deploy-helper" + assert results["b"] is None # stale base under the lock -> refused + assert loader.get_auto_skill_version("auto/deploy-helper") == 2 # ONE bump + live_body = (loader._dir / "auto" / "deploy-helper" / "SKILL.md").read_text( + encoding="utf-8" + ) + assert "the new way" in live_body and "the B way" not in live_body + # Loser's candidate survives for review; winner's was consumed. + assert _pending_slugs(loader) == ["upd-b"] + + +def test_stage_token_mismatch_inside_lock_refuses_promotion(loader): + """A candidate swapped between the caller's pre-lock ownership check and + the locked promotion (concurrent dismiss + same-slug re-stage) must be + refused INSIDE the lock: the pending meta's ``stage_token`` no longer + matches the staging flow's token, so the unreviewed replacement stays + pending instead of going live.""" + import secrets + + _write_live(loader, "deploy-helper", version=1) + # Production-shaped tokens (secrets.token_hex(16)): pins that the pending + # meta redaction pass does NOT scrub the token, which would make every + # unattended promotion self-refuse. + tok_original = secrets.token_hex(16) + tok_impostor = secrets.token_hex(16) + assert ( + loader.stage_skill_candidate( + "upd-swap", + description="updated", + triggers="t", + procedure_md=_NEW_STEPS, + provenance=_prov(), + scripts=None, + kind="update", + target="auto/deploy-helper", + base_version=1, + notify=False, + stage_token=tok_original, + ) + == "auto/upd-swap" + ) + # Simulate the swap: the original candidate is dismissed and a different + # one is re-staged under the SAME slug with its own token. + loader.dismiss_pending_skill("upd-swap") + assert ( + loader.stage_skill_candidate( + "upd-swap", + description="updated", + triggers="t", + procedure_md="## Steps\n\n1. the impostor way\n", + provenance=_prov(), + scripts=None, + kind="update", + target="auto/deploy-helper", + base_version=1, + notify=False, + stage_token=tok_impostor, + ) + == "auto/upd-swap" + ) + # The staging flow still holds the ORIGINAL token — promotion must refuse. + assert ( + loader.approve_pending_update( + "upd-swap", refuse_scripts=True, expected_stage_token=tok_original + ) + is None + ) + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 # untouched + assert _pending_slugs(loader) == ["upd-swap"] # impostor stays reviewable + # Matching token (the impostor's own flow) still promotes normally. + assert ( + loader.approve_pending_update( + "upd-swap", refuse_scripts=True, expected_stage_token=tok_impostor + ) + == "auto/deploy-helper" + ) + assert loader.get_auto_skill_version("auto/deploy-helper") == 2 + + +@needs_flock +def test_script_injection_after_lock_acquisition_refused(loader, monkeypatch): + """A scripts/ dir injected AFTER the promotion lock is acquired (and after + the entry precondition) must still be refused: the re-check runs inside + the lock, and under ``refuse_scripts`` the copy path is unreachable.""" + _write_live(loader, "deploy-helper", version=1) + _stage_candidate(loader, notify=False) + pend = loader._pending_root() / "deploy-helper-update" + lock_states: list[str] = [] + real_redact = loader._validate_and_redact_candidate + + def _probe_then_inject(src, target_name): + # Prove the per-target lock is already held at this point: a + # non-blocking flock on a second fd must fail with EWOULDBLOCK. + lock_path = loader._locks_root() / "deploy-helper.lock" + fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o600) + try: + try: + flock_compat.flock(fd, flock_compat.LOCK_EX | flock_compat.LOCK_NB) + lock_states.append("unlocked") + flock_compat.flock(fd, flock_compat.LOCK_UN) + except OSError: + lock_states.append("locked") + finally: + os.close(fd) + (pend / "scripts").mkdir(exist_ok=True) + (pend / "scripts" / "sneaky.py").write_text("print('hi')\n", encoding="utf-8") + return real_redact(src, target_name) + + monkeypatch.setattr(loader, "_validate_and_redact_candidate", _probe_then_inject) + + assert loader.auto_apply_pending_update("deploy-helper-update") is None + + assert lock_states == ["locked"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + assert not (loader._dir / "auto" / "deploy-helper" / "scripts" / "sneaky.py").exists() + assert _pending_slugs(loader) == ["deploy-helper-update"] + + +@needs_flock +def test_promotion_refused_while_lock_held_elsewhere(loader, monkeypatch): + """A held per-target lock (e.g. a promotion in another process) makes a + contending promotion FAIL SAFE after the bounded poll: refused, candidate + left pending, live untouched. A crashed holder cannot cause this state to + persist — flock is released by the kernel on process death.""" + _write_live(loader, "deploy-helper", version=1) + _stage_candidate(loader, notify=False) + monkeypatch.setattr(S, "_PROMOTE_LOCK_TIMEOUT_S", 0.2) + lock_path = loader._locks_root() / "deploy-helper.lock" + lock_path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o600) + flock_compat.flock(fd, flock_compat.LOCK_EX) + try: + assert loader.approve_pending_update("deploy-helper-update") is None + finally: + flock_compat.flock(fd, flock_compat.LOCK_UN) + os.close(fd) + assert _pending_slugs(loader) == ["deploy-helper-update"] + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + + +# ── pending-lookup failure re-fires the suppressed staging notification ── + + +def test_pending_lookup_failure_refires_staged_notification(loader, monkeypatch): + """A transient ``get_pending_skill`` failure during the ownership check + must not leave the just-staged candidate invisible: the review request + that staging suppressed (notify=False) is re-fired, and the refusal is + audited as ``pending_lookup_failed`` (distinct from ownership_mismatch).""" + _write_live(loader, "deploy-helper", version=1) + staged_seen: list[dict] = [] + S.set_pending_staged_hook(staged_seen.append) + c = _mk(loader, approval_required=False) + + def _boom(slug): + raise RuntimeError("transient pending-store failure") + + monkeypatch.setattr(loader, "get_pending_skill", _boom) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + _stage_update(c) + finally: + ctx.stop() + + # Candidate still pending, live untouched. + pending = _pending_slugs(loader) + assert len(pending) == 1 + assert loader.get_auto_skill_version("auto/deploy-helper") == 1 + # The suppressed staging notification fired exactly once, for our slug. + assert [p["slug"] for p in staged_seen] == pending + # Audited with the lookup-failure reason. + reasons = [ + r["metadata"].get("reason") + for r in recorded + if r.get("outcome") == "auto_apply_failed" + ] + assert reasons == ["pending_lookup_failed"] diff --git a/test/test_skill_update_flow.py b/test/test_skill_update_flow.py index 5665c33040a..f5f0a6251f5 100644 --- a/test/test_skill_update_flow.py +++ b/test/test_skill_update_flow.py @@ -33,6 +33,9 @@ def __init__(self, *, existing=None, live_body=_LIVE_BODY, version=3, find=None) self._version = version self._find = find self.staged: list[dict] = [] + self.auto_applied: list[str] = [] + self.auto_apply_result = ("auto/deploy-helper", 4) + self.re_notified: list[str] = [] # dedupe inputs def list_auto_skills(self): @@ -64,6 +67,8 @@ def stage_skill_candidate( base_version=None, scripts=None, source="consolidation", + notify=True, + stage_token=None, ): self.staged.append( { @@ -75,10 +80,38 @@ def stage_skill_candidate( "target": target, "base_version": base_version, "scripts": scripts, + "notify": notify, + "created_at": provenance.created_at, + "stage_token": stage_token, } ) return f"auto/{slug}" + # auto-apply surface (used by the approval-off flow) + def get_pending_skill(self, slug): + st = [s for s in self.staged if s["slug"] == slug] + if not st: + return None + return { + "slug": slug, + "kind": st[-1]["kind"], + "meta": { + "created_at": st[-1]["created_at"], + "stage_token": st[-1]["stage_token"], + }, + } + + def auto_apply_pending_update(self, slug, *, expected_stage_token=None): + self.auto_applied.append(slug) + self.auto_apply_tokens = getattr(self, "auto_apply_tokens", []) + self.auto_apply_tokens.append(expected_stage_token) + if self.auto_apply_result == "raise": + raise RuntimeError("boom") + return self.auto_apply_result + + def emit_pending_staged(self, slug): + self.re_notified.append(slug) + # new-path fallbacks (unused in these tests but referenced by the code) def create_auto_skill(self, *a, **k): return None @@ -87,13 +120,13 @@ def run_skill_lifecycle(self, *a, **k): return None -def _mk(loader, **kw): +def _mk(loader, *, approval_required=True, **kw): return HistoryConsolidator( log=MagicMock(), memory=MagicMock(), skills_loader=loader, auto_skills_enabled=True, - approval_required=True, + approval_required=approval_required, **kw, ) @@ -938,3 +971,193 @@ async def fake_merge(*a): # version live drifted to during the merge (2). The staleness guard will then # correctly refuse this candidate instead of silently applying it over v2. assert loader.staged[0]["base_version"] == 1 + + +# ── Auto-apply routing (approval disabled) ─────────────────────────────────── + + +def test_stage_update_flag_off_prose_only_auto_applies(): + """With approval off and no scripts, the staged update is promoted through + the loader's auto-apply path and the review notification is suppressed.""" + loader = FakeLoader() + c = _mk(loader, approval_required=False) + c._event_loop = None + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + ) + finally: + ctx.stop() + + assert loader.staged[0]["notify"] is False + assert loader.auto_applied == ["deploy-helper-update"] + assert loader.re_notified == [] + ev = [r for r in recorded if r.get("outcome") == "auto_applied_update"] + assert ev and ev[0]["metadata"] == { + "name": "auto/deploy-helper-update", + "target": "auto/deploy-helper", + "new_version": 4, + } + # The stage itself is still audited. + assert [r for r in recorded if r.get("outcome") == "staged_update"] + + +def test_stage_update_flag_off_with_scripts_does_not_auto_apply(): + loader = FakeLoader() + c = _mk(loader, approval_required=False) + c._event_loop = None + ctx = _sel_recorder([]) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + scripts=[{"filename": "go.py", "content": "print('hi')\n"}], + scripts_supplied=True, + ) + finally: + ctx.stop() + + assert loader.staged[0]["notify"] is True + assert loader.auto_applied == [] + + +def test_stage_update_flag_off_rejected_scripts_does_not_auto_apply(): + """A candidate that SUPPLIED scripts must never auto-apply, even when the + validator rejected all of them (scripts=None but scripts_supplied=True).""" + loader = FakeLoader() + c = _mk(loader, approval_required=False) + c._event_loop = None + ctx = _sel_recorder([]) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + scripts=None, + scripts_supplied=True, + ) + finally: + ctx.stop() + + assert loader.staged[0]["notify"] is True + assert loader.auto_applied == [] + + +def test_stage_update_flag_on_does_not_auto_apply(): + loader = FakeLoader() + c = _mk(loader, approval_required=True) + c._event_loop = None + ctx = _sel_recorder([]) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + ) + finally: + ctx.stop() + + assert loader.staged[0]["notify"] is True + assert loader.auto_applied == [] + + +def test_stage_update_auto_apply_failure_re_fires_review_notification(): + """A refused promotion fails SAFE: the candidate stays pending, so the + review request suppressed at staging time is re-fired.""" + loader = FakeLoader() + loader.auto_apply_result = None + c = _mk(loader, approval_required=False) + c._event_loop = None + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + ) + finally: + ctx.stop() + + assert loader.auto_applied == ["deploy-helper-update"] + assert loader.re_notified == ["deploy-helper-update"] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + + +def test_stage_update_auto_apply_exception_re_fires_review_notification(): + loader = FakeLoader() + loader.auto_apply_result = "raise" + c = _mk(loader, approval_required=False) + c._event_loop = None + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + ) + finally: + ctx.stop() + + assert loader.re_notified == ["deploy-helper-update"] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"] + + +def test_stage_update_auto_apply_skips_foreign_pending_candidate(monkeypatch): + """When the collision family is exhausted, stage_skill_candidate returns the + slug name WITHOUT staging — the pending entry under that slug is a + DIFFERENT, previously-staged candidate. Auto-apply must not promote it.""" + loader = FakeLoader() + c = _mk(loader, approval_required=False) + c._event_loop = None + # The pending entry is foreign: even an identical same-second created_at + # must not count as ownership — only the random stage_token written by + # THIS flow's staging call does, and a foreign entry carries a different + # one (or none, for a deferred re-stage that wrote nothing). + monkeypatch.setattr( + loader, + "get_pending_skill", + lambda slug: { + "slug": slug, + "kind": "update", + "meta": { + "created_at": "1999-01-01T00:00:00+00:00", + "stage_token": "someone-elses-token", + }, + }, + ) + recorded: list[dict] = [] + ctx = _sel_recorder(recorded) + try: + c._stage_skill_update( + key="sess", + target_key="auto/deploy-helper", + description="d", + triggers="t", + procedure_md="## Steps\n1. cand\n", + ) + finally: + ctx.stop() + + assert loader.auto_applied == [] + # Nothing of ours is pending, so no re-notification either. + assert loader.re_notified == [] + assert not [r for r in recorded if r.get("outcome") == "auto_applied_update"]