tests: pin ink engine in _make_tui_argv npm-bootstrap tests (post-merge semantic fix)

Main's rewritten test_tui_npm_install.py tests call _make_tui_argv expecting
the Ink/npm flow unconditionally; with the dual-engine dispatch merged in,
_resolve_tui_engine() auto-selects opentui whenever ui-opentui/dist is built
in the repo, routing the call away from the path under test (first subprocess
became 'node --version' instead of 'npm run build'). Pin the engine to ink
via an autouse fixture, mirroring the existing pinning precedent in
test_tui_resume_flow.py.
This commit is contained in:
alt-glitch 2026-06-12 10:32:40 +05:30
parent ab37440ce6
commit e1067dbbe5
756 changed files with 79874 additions and 19585 deletions

View file

@ -822,6 +822,75 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
# Main entry point
# =============================================================================
# ContextVar bypass: set while replaying an already-approved staged skill write
# so skill_manage() does not re-gate (and re-stage) it.
import contextvars as _ctxvars
_skill_gate_bypass: "_ctxvars.ContextVar[bool]" = _ctxvars.ContextVar(
"skill_gate_bypass", default=False
)
def _apply_skill_write_gate(action, name, **payload_kwargs):
"""Evaluate the skill write gate. Returns a JSON tool-result string when the
write should NOT proceed (blocked or staged), or None to perform the real
write. Bypassed during approved-pending replay.
"""
if action not in {"create", "edit", "patch", "delete", "write_file", "remove_file"}:
return None
if _skill_gate_bypass.get():
return None
try:
from tools import write_approval as wa
except Exception:
return None # fail open
decision = wa.evaluate_gate(wa.SKILLS)
if decision.allow:
return None
if decision.blocked:
return tool_error(decision.message, success=False)
# stage — record the full skill_manage kwargs so approval can replay it.
payload = {"action": action, "name": name}
payload.update({k: v for k, v in payload_kwargs.items() if v is not None})
gist = wa.skill_gist(
action, name,
content=payload_kwargs.get("content") or "",
file_path=payload_kwargs.get("file_path") or "",
old_string=payload_kwargs.get("old_string") or "",
new_string=payload_kwargs.get("new_string") or "",
)
record = wa.stage_write(wa.SKILLS, payload, summary=gist, origin=wa.current_origin())
return json.dumps(
{"success": True, "staged": True, "pending_id": record["id"],
"gist": gist, "message": decision.message},
ensure_ascii=False,
)
def apply_skill_pending(payload: Dict[str, Any]) -> str:
"""Replay a staged skill write, bypassing the gate. Returns the tool result
JSON string. Called by the /skills approve handler.
"""
token = _skill_gate_bypass.set(True)
try:
return skill_manage(
action=payload.get("action", ""),
name=payload.get("name", ""),
content=payload.get("content"),
category=payload.get("category"),
file_path=payload.get("file_path"),
file_content=payload.get("file_content"),
old_string=payload.get("old_string"),
new_string=payload.get("new_string"),
replace_all=payload.get("replace_all", False),
absorbed_into=payload.get("absorbed_into"),
)
finally:
_skill_gate_bypass.reset(token)
def skill_manage(
action: str,
name: str,
@ -839,6 +908,19 @@ def skill_manage(
Returns JSON string with results.
"""
# Approval gate: when on, stages the write for review (skills are too large
# to review inline, so they always stage regardless of origin); when off
# (default) passes straight through. The gate is bypassed when this call is
# itself replaying an already-approved staged write (_skill_apply_pending).
gate_result = _apply_skill_write_gate(
action, name, content=content, category=category,
file_path=file_path, file_content=file_content,
old_string=old_string, new_string=new_string,
replace_all=replace_all, absorbed_into=absorbed_into,
)
if gate_result is not None:
return gate_result
if action == "create":
if not content:
return tool_error("content is required for 'create'. Provide the full SKILL.md text (frontmatter + body).", success=False)