Finish Agent Skill installs from trusted runtime receipts

.github/ci/suites.v1.json

@@ -415,7 +415,9 @@
         "tests/test_gateway/test_skill_install_status.py",
         "tests/test_skills/test_hub_install_operations.py",
         "tests/test_skills/test_staging_io_worker.py",
-        "src/opensquilla/engine/agent.py"
+        "src/opensquilla/engine/agent.py",
+        "tests/test_engine/test_skill_install_turn.py",
+        "src/opensquilla/engine/runtime.py"
       ]
     },
     "managed-toolchain": {

.github/scripts/plan_ci.py

@@ -168,6 +168,7 @@
         "tests/test_skills_hub_lockfile_contract.py",
         "tests/test_skills_hub_doctor.py",
         "tests/test_skills_hash_consumers.py",
+        "tests/test_engine/test_skill_install_turn.py",
         "tests/test_engine/test_skill_install_settlement.py",
         "tests/test_gateway/test_skill_install_status.py",
         "tests/test_skills/test_hub_install_operations.py",
@@ -207,6 +208,7 @@
         "src/opensquilla/cli/skills_meta_cmd.py",
         "src/opensquilla/application/skill_catalog.py",
         "src/opensquilla/application/skill_management.py",
+        "src/opensquilla/engine/runtime.py",
         "src/opensquilla/engine/agent.py",
         "src/opensquilla/application/skill_source.py",
         "src/opensquilla/application/skill_proposal_review.py",

.github/scripts/windows_test_assignments.json

@@ -1369,6 +1369,7 @@
       "tests/test_engine/test_runtime_meta_invoke_surfacing.py",
       "tests/test_engine/test_runtime_tool_run_budget.py",
       "tests/test_engine/test_skill_install_settlement.py",
+      "tests/test_engine/test_skill_install_turn.py",
       "tests/test_engine/test_spawn_depth_unified.py",
       "tests/test_engine/test_stream_repetition_guard.py",
       "tests/test_engine/test_stream_wrappers.py",

.github/scripts/windows_test_durations.json

@@ -1597,6 +1597,7 @@
     "tests/test_engine/test_skill_install_settlement.py": 0.01,
     "tests/test_gateway/test_skill_install_status.py": 0.01,
     "tests/test_skills/test_hub_install_operations.py": 0.01,
-    "tests/test_skills/test_staging_io_worker.py": 0.01
+    "tests/test_skills/test_staging_io_worker.py": 0.01,
+    "tests/test_engine/test_skill_install_turn.py": 0.01
   }
 }

.github/workflows/ci.yml

@@ -2057,6 +2057,7 @@ jobs:
             tests/test_skills_hub_doctor.py \
             tests/test_skills_hash_consumers.py \
             tests/test_skill_install_source.py \
+            tests/test_engine/test_skill_install_turn.py \
             tests/test_engine/test_skill_install_settlement.py \
             tests/test_gateway/test_skill_install_status.py \
             tests/test_skills/test_hub_install_operations.py \

src/opensquilla/engine/agent.py

@@ -6041,6 +6041,14 @@ async def _turn_generator(
         self.config.metadata.pop("reasoning_replay_context_rebuilt", None)
         self._provider_tool_result_overrides = {}
         self._current_turn_message = message
+        from opensquilla.skills.install_turn import SkillInstallTurn
+
+        install_turn = SkillInstallTurn(semantic_message or message)
+        if self._tool_context is not None:
+            self._tool_context.skill_install_turn = install_turn
+            install_turn.finalization_allowed = not (
+                self._tool_context.goal_context or self._tool_context.plan_run_id
+            )
         _meta_invoke_turn_count.set(0)
         usage_scope = current_usage_accounting_scope()
         reasoning_block_index = 0
@@ -6900,6 +6908,7 @@ async def _claim_pending_inputs_for_next_call() -> bool:
             )
             turn_messages.append(staged_pending_input_message)
             pending_input_batch_staged = True
+            install_turn.accept_user_input()
             return True
 
         async def _mark_staged_pending_inputs_applied(
@@ -11906,6 +11915,22 @@ async def _collect_tool_tasks(
                         finally:
                             for task in pending:
                                 tc = task_to_tool_call[task]
+                                if tc.tool_name == "skill_install_community":
+                                    from opensquilla.skills.install_source import (
+                                        resolve_install_source,
+                                    )
+
+                                    identifier = str(tc.arguments.get("identifier", ""))
+                                    source = resolve_install_source(
+                                        identifier, tc.arguments.get("source"),
+                                    )
+                                    receipt = install_turn.previous(identifier, source)
+                                    if receipt is not None:
+                                        results_by_id[tc.tool_use_id] = ToolResult(
+                                            tool_use_id=tc.tool_use_id,
+                                            tool_name=tc.tool_name,
+                                            content=json.dumps(receipt),
+                                        )
                                 result = results_by_id.get(tc.tool_use_id)
                                 if (
                                     result is None
@@ -12044,6 +12069,9 @@ async def _run_after_key_lock() -> ToolResult:
                         yield event
 
                 for tc in tool_calls:
+                    peek_installs = getattr(pending_input_provider, "peek_pending", None)
+                    if callable(peek_installs) and peek_installs():
+                        install_turn.finalization_allowed = False
                     if dispatch_boundary is not None:
                         results_by_id[tc.tool_use_id] = _not_executed_after_dispatch_boundary(
                             tc,
@@ -12150,6 +12178,8 @@ async def _run_after_key_lock() -> ToolResult:
                             # with a not-executed result, then perform exactly
                             # one tool-free final-summary model call.
                             dispatch_boundary = mutex_result
+                        if mutex_result is not None and install_turn.complete:
+                            dispatch_boundary = mutex_result
                         if _plan_run_checkpoint_enters_delivery_phase(mutex_result):
                             plan_run_delivery_only = True
 
@@ -12540,6 +12570,20 @@ async def _run_after_key_lock() -> ToolResult:
                         failure_anchor_summary=failure_anchor_summary,
                     ):
                         append_runtime_event(self.config.runtime_events_path, runtime_event)
+                if install_turn.complete:
+                    await _claim_pending_inputs_for_next_call()
+                    peek_installs = getattr(pending_input_provider, "peek_pending", None)
+                    if callable(peek_installs) and peek_installs():
+                        install_turn.finalization_allowed = False
+                if install_turn.complete:
+                    final_response_text = install_turn.final_text()
+                    final_text_parts[:] = [final_response_text]
+                    applied_model_call_boundaries.clear()
+                    yield TextDeltaEvent(
+                        text=final_response_text, presentation="answer",
+                        generation_epoch=generation_epoch,
+                    )
+                    break
                 budget_error = (
                     None if accepted_goal_terminal_status is not None else _turn_budget_error()
                 )
@@ -12573,6 +12617,9 @@ async def _run_after_key_lock() -> ToolResult:
                     yield self._transition(AgentState.THINKING)
                     continue
                 await _claim_pending_inputs_for_next_call()
+                peek_installs = getattr(pending_input_provider, "peek_pending", None)
+                if callable(peek_installs) and peek_installs():
+                    install_turn.finalization_allowed = False
                 if terminal_projection_preflight_error:
                     self._write_turn_call_log(
                         "tool_argument_projection_rehydrate_recovery",

src/opensquilla/engine/runtime.py

@@ -6448,6 +6448,13 @@ async def _load_turn_transcript() -> Sequence[Any]:
                 image_failure_cleanup = temporary_images.cleanup
                 attachment_cleanup = image_failure_cleanup
 
+            if tool_context is not None:
+                from opensquilla.skills.install_turn import SkillInstallTurn
+
+                tool_context = replace(
+                    tool_context,
+                    skill_install_turn=SkillInstallTurn(semantic_message or message),
+                )
             pt_outcome = await self._provider_and_tools_stage.run(
                 ProviderAndToolsStageInput(
                     session_key=session_key,
@@ -8873,6 +8880,20 @@ def _build_tools(
             ctx is not None and str(getattr(ctx, "plan_run_id", "") or "").strip()
         )
         if ctx is not None:
+            from opensquilla.skills.catalog_policy import project_public_catalog
+            from opensquilla.skills.install_turn import SkillInstallTurn
+
+            skill_tools: set[str] = set()
+            if isinstance(ctx.skill_install_turn, SkillInstallTurn):
+                skill_tools.update(ctx.skill_install_turn.surface_tools())
+            if project_public_catalog(
+                loaded_skills, coding_mode=ctx.coding_mode, include_stable_meta=False,
+            ):
+                skill_tools.update({"skill_list", "skill_view"})
+            if skill_tools:
+                if ctx.surfaced_tools is None:
+                    ctx.surfaced_tools = set()
+                ctx.surfaced_tools.update(skill_tools)
             # A lossy tool-result projection is only useful when the model can
             # recover the stored original. Surface the read-only retrieval tool
             # before the first schema is built; normal allow/deny/profile policy

src/opensquilla/skills/install_turn.py

@@ -0,0 +1,147 @@
+"""Trusted, turn-local installation receipts and conservative request scope."""
+
+from __future__ import annotations
+
+import copy
+import re
+from dataclasses import dataclass, field
+from typing import Any
+
+from opensquilla.skills.install_source import resolve_install_source
+
+_INSTALL = re.compile(r"\binstall(?:ation)?\b|安装|装上|装一下", re.I)
+_SEARCH = re.compile(r"\b(?:find|search|browse|locate)\b|搜索|查找|找.*技能", re.I)
+_PREFIX = re.compile(
+    r"^(?:(?:please|can you|could you|help me|I want you to)\s+|请|帮我|麻烦你|帮忙)*"
+    r"(?:install\b|安装|装上|装一下)\s*", re.I,
+)
+_URL = re.compile(r"(?:https?://|github\.com/)[^\s<>\[\](),，、；;。]+", re.I)
+_MIXED = re.compile(
+    r"\b(?:then|use|using|run|write|create|verify|check|explain|test|update|remove)\b"
+    r"|然后|并且|使用|写|运行|验证|检查|解释|测试|更新|删除", re.I,
+)
+_SCAFFOLD = re.compile(
+    r"\b(?:the|these|this|following|skills?|and|from|for me)\b"
+    r"|以下|这些|这个|这两个|这几个|技能|和|以及|并", re.I,
+)
+
+
+def _key(identifier: str, source: str | None = None) -> tuple[str, str]:
+    return resolve_install_source(identifier, source), identifier.strip().rstrip("/")
+
+
+def install_targets(request: str) -> tuple[tuple[str, str], ...]:
+    """Accept only an install imperative followed by an exact list of targets."""
+    text = request.strip()
+    if "```" in text or re.search(r"(?m)^\s*>|<[^>]+>", text):
+        return ()
+    prefix = _PREFIX.match(text)
+    if prefix is None:
+        return ()
+    text = text[prefix.end():]
+    # Markdown labels are presentation; the link target is the requested reference.
+    text = re.sub(r"\[[^\]\n]+\]\(([^)\n]+)\)", r"\1", text)
+    urls = _URL.findall(text)
+    remaining = _URL.sub(" ", text)
+    if _MIXED.search(remaining):
+        return ()
+    remaining = _SCAFFOLD.sub(" ", remaining)
+    tokens = [part for part in re.split(r"[\s,，、;；:：.!！?？。`*]+", remaining) if part]
+    tokens = [part for part in tokens if part != "-"]
+    if any(not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_./@-]*", part) for part in tokens):
+        return ()
+    if any(resolve_install_source(url) != "github" for url in urls):
+        return ()
+    targets = [_key(value) for value in [*urls, *tokens]]
+    return tuple(dict.fromkeys(targets)) if 0 < len(targets) <= 100 else ()
+
+
+@dataclass
+class SkillInstallTurn:
+    request: str
+    receipts: dict[tuple[str, str], dict[str, Any]] = field(default_factory=dict)
+    targets: tuple[tuple[str, str], ...] = field(init=False)
+    finalization_allowed: bool = True
+
+    def __post_init__(self) -> None:
+        self.targets = install_targets(self.request)
+
+    def surface_tools(self) -> set[str]:
+        result: set[str] = set()
+        if _INSTALL.search(self.request):
+            result.add("skill_install_community")
+        if _SEARCH.search(self.request) and re.search(r"skills?|技能", self.request, re.I):
+            result.add("skill_search_community")
+        return result
+
+    def accept_user_input(self) -> None:
+        """Reconsider failures only at the runtime's trusted input boundary.
+
+        A new request may select a directory or authorize a retry. It does not
+        itself acknowledge scanner risk; the installer still validates the
+        supplied confirmation. Successful installs remain reusable.
+        """
+        self.finalization_allowed = False
+        self.receipts = {
+            key: receipt for key, receipt in self.receipts.items()
+            if receipt.get("success") is True
+        }
+
+    def previous(self, identifier: str, source: str) -> dict[str, Any] | None:
+        result = self.receipts.get(_key(identifier, source))
+        if result is not None:
+            return copy.deepcopy(result)
+        # A directory-choice response grants no authority to choose a directory.
+        # Only a new user turn or accepted user input may reconsider this result.
+        for receipt in self.receipts.values():
+            for diagnostic in receipt.get("diagnostics", []):
+                if diagnostic.get("code") != "SOURCE_TREE_AMBIGUOUS":
+                    continue
+                for candidate in diagnostic.get("details", {}).get("candidates", []):
+                    if candidate.get("identifier") == identifier:
+                        return copy.deepcopy(receipt)
+        return None
+
+    def record(self, identifier: str, source: str, result: dict[str, Any]) -> None:
+        self.receipts[_key(identifier, source)] = copy.deepcopy(result)
+
+    @property
+    def complete(self) -> bool:
+        return bool(self.finalization_allowed and self.targets) and all(
+            self.receipts.get(target, {}).get("success") is True for target in self.targets
+        )
+
+    def final_text(self) -> str:
+        chinese = bool(re.search(r"[\u3400-\u9fff]", self.request))
+        lines = []
+        for target in self.targets:
+            receipt = self.receipts[target]
+            name = str(receipt.get("name") or target[1])
+            usable = receipt.get("instruction_usable") is True
+            lifecycle = receipt.get("lifecycle") or {}
+            if chinese:
+                status = "已安装，下一回合可用" if usable else "已安装，当前尚不可用"
+            else:
+                status = "Installed; available next turn" if usable else "Installed; not yet usable"
+            degraded = lifecycle.get("compatibility_state") == "degraded"
+            if degraded:
+                status += "；兼容性受限" if chinese else "; limited compatibility"
+            if not usable:
+                states = [
+                    str(lifecycle.get(key, ""))
+                    for key in ("load_state", "selection_state", "readiness_state")
+                ]
+                states = [value for value in states if value]
+                if states:
+                    status += " (" + ", ".join(states) + ")"
+            line = f"{name}: {status}."
+            if not usable or degraded:
+                hints = list(dict.fromkeys(
+                    diagnostic["hint"].strip()
+                    for diagnostic in receipt.get("diagnostics", [])
+                    if isinstance(diagnostic.get("hint"), str) and diagnostic["hint"].strip()
+                ))
+                if hints:
+                    line += " " + " ".join(hints[:3])
+            lines.append(line)
+        return "\n".join(lines)