ZacharyZhang-NY
e37418c489
Land the crash-recoverable managed-file transaction engine (journal, staging, commit, restart recovery, opaque snapshots, private artifacts, Windows DACL hardening) and the Kiro IDE 1.0 workspace hook plugin with legacy migration. Repair five defects that kept the suite red (426 failures to zero): - Anchor the transaction state directory at ~/.elydora/transactions instead of the shared system temp directory. The recovery journal now shares fate with the runtimes it repairs, survives temp cleaners, and isolates naturally under per-test fake homes. - Skip pending-transaction recovery when no state directory exists so rejected installs never create runtime directories. - Create the runtime root's parent before hardening it, matching the Go SDK's expand behavior for fresh homes. - Stop using file timestamps as identity evidence: Windows exposes mtime/ctime with delayed cross-handle visibility. Reads now prove stability through a verification re-read; snapshots compare device, inode, size, and SHA-256. - Stop comparing synthetic Windows permission bits across stat boundaries: path stats derive execute bits from file extensions while handle stats do not, which falsely flagged every managed .cmd wrapper. Snapshots store raw modes for restoration and compare portable bits. Reservation cleanup now verifies content before discarding so a concurrent writer's file at a rollback path is preserved, and the install-failure tests assert the precise contract: no agent runtime residue, with the durable state directory allowed to remain. Gates: pytest 807 passed, mypy clean, pip check clean, wheel builds.
275 lines
8.9 KiB
Python
275 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess # nosec B404
|
|
import sys
|
|
from typing import Any, Dict
|
|
|
|
import pytest
|
|
|
|
from elydora.plugins import _transaction, droid
|
|
from elydora.plugins.droid_config import render_document, source_documents
|
|
from elydora.plugins.droid_installation import (
|
|
commit_droid_installation,
|
|
commit_droid_uninstall,
|
|
preflight_droid_installation,
|
|
prepare_droid_installation,
|
|
prepare_droid_uninstall,
|
|
)
|
|
from elydora.plugins.droid_io import read_sources
|
|
|
|
from droid_support import (
|
|
AGENT_ID,
|
|
VALID_PRIVATE_KEY,
|
|
DroidFixture,
|
|
assert_no_transaction_files,
|
|
assert_snapshot,
|
|
load_jsonc,
|
|
snapshot,
|
|
write_json,
|
|
)
|
|
|
|
|
|
def _prepare_installation(
|
|
fixture: DroidFixture,
|
|
**overrides: str,
|
|
):
|
|
config: Dict[str, Any] = dict(fixture.config)
|
|
config.update(overrides)
|
|
sources = read_sources()
|
|
paths = preflight_droid_installation(config, sources)
|
|
rendered = droid._render_installation( # noqa: SLF001
|
|
sources,
|
|
paths.guard_path,
|
|
paths.audit_path,
|
|
)
|
|
return prepare_droid_installation(config, sources, rendered)
|
|
|
|
|
|
def _managed_paths(fixture: DroidFixture) -> tuple[Path, ...]:
|
|
return (
|
|
fixture.guard_path,
|
|
fixture.runtime_config,
|
|
fixture.private_key_path,
|
|
fixture.audit_path,
|
|
fixture.root_path,
|
|
)
|
|
|
|
|
|
def test_install_rolls_back_runtime_after_late_hook_commit_failure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(monkeypatch, tmp_path)
|
|
prepared = _prepare_installation(fixture)
|
|
real_replace = _transaction._replace_physical
|
|
failed = False
|
|
|
|
def fail_hook_commit(
|
|
source: str,
|
|
destination: str,
|
|
directory: _transaction.PinnedDirectory,
|
|
) -> None:
|
|
nonlocal failed
|
|
if not failed and Path(destination) == fixture.root_path:
|
|
failed = True
|
|
raise OSError("injected Droid hook commit failure")
|
|
real_replace(source, destination, directory)
|
|
|
|
monkeypatch.setattr(_transaction, "_replace_physical", fail_hook_commit)
|
|
with pytest.raises(OSError, match="injected Droid hook commit failure"):
|
|
commit_droid_installation(prepared)
|
|
assert failed is True
|
|
for path in _managed_paths(fixture):
|
|
assert not path.exists()
|
|
assert_no_transaction_files(tmp_path)
|
|
|
|
|
|
def test_concurrent_active_hook_replacement_prevents_runtime_commit(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
before = snapshot(_managed_paths(fixture))
|
|
prepared = _prepare_installation(fixture, org_id="org-updated")
|
|
concurrent = '{"hooks":{"PreToolUse":[]},"owner":"concurrent"}\n'
|
|
fixture.root_path.write_text(concurrent, encoding="utf-8")
|
|
with pytest.raises(OSError, match="hooks changed during Install Factory Droid"):
|
|
commit_droid_installation(prepared)
|
|
for path, source in before.items():
|
|
if path != fixture.root_path:
|
|
assert path.read_text(encoding="utf-8") == source
|
|
assert fixture.root_path.read_text(encoding="utf-8") == concurrent
|
|
assert_no_transaction_files(tmp_path)
|
|
|
|
|
|
def test_concurrent_inactive_local_source_creation_is_detected(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
before = snapshot(_managed_paths(fixture))
|
|
prepared = _prepare_installation(fixture, org_id="org-updated")
|
|
concurrent = '{"hooksDisabled":true}\n'
|
|
fixture.local_settings_path.write_text(concurrent, encoding="utf-8")
|
|
with pytest.raises(OSError, match="local settings changed during Install"):
|
|
commit_droid_installation(prepared)
|
|
assert_snapshot(before)
|
|
assert fixture.local_settings_path.read_text(encoding="utf-8") == concurrent
|
|
assert_no_transaction_files(tmp_path)
|
|
|
|
|
|
def test_concurrent_project_policy_change_is_detected(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(
|
|
monkeypatch,
|
|
tmp_path,
|
|
project_settings={"hooksDisabled": False},
|
|
)
|
|
fixture.install()
|
|
before = snapshot(_managed_paths(fixture))
|
|
prepared = _prepare_installation(fixture, org_id="org-updated")
|
|
concurrent = '{"hooksDisabled":true}\n'
|
|
fixture.project_settings_path.write_text(concurrent, encoding="utf-8")
|
|
with pytest.raises(OSError, match="project settings changed during Install"):
|
|
commit_droid_installation(prepared)
|
|
assert_snapshot(before)
|
|
assert fixture.project_settings_path.read_text(encoding="utf-8") == concurrent
|
|
assert_no_transaction_files(tmp_path)
|
|
|
|
|
|
def test_same_content_replacement_is_rejected_by_physical_identity(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
before = snapshot(_managed_paths(fixture))
|
|
sources = read_sources()
|
|
original = fixture.root_path.read_text(encoding="utf-8")
|
|
fixture.root_path.unlink()
|
|
fixture.root_path.write_text(original, encoding="utf-8")
|
|
config: Dict[str, Any] = dict(fixture.config)
|
|
config["org_id"] = "org-updated"
|
|
paths = preflight_droid_installation(config, sources)
|
|
rendered = droid._render_installation( # noqa: SLF001
|
|
sources,
|
|
paths.guard_path,
|
|
paths.audit_path,
|
|
)
|
|
prepared = prepare_droid_installation(config, sources, rendered)
|
|
with pytest.raises(OSError, match="hooks changed during Install"):
|
|
commit_droid_installation(prepared)
|
|
assert fixture.root_path.read_text(encoding="utf-8") == original
|
|
for path, source in before.items():
|
|
if path != fixture.root_path:
|
|
assert path.read_text(encoding="utf-8") == source
|
|
assert_no_transaction_files(tmp_path)
|
|
|
|
|
|
def test_orphaned_runtime_is_preserved_without_verifiable_identity(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(monkeypatch, tmp_path)
|
|
fixture.agent_dir.mkdir(parents=True)
|
|
fixture.guard_path.write_text("orphaned guard\n", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="identity cannot be verified"):
|
|
fixture.install()
|
|
assert fixture.guard_path.read_text(encoding="utf-8") == "orphaned guard\n"
|
|
assert not fixture.root_path.exists()
|
|
|
|
|
|
def test_uninstall_rolls_back_all_hook_documents(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = DroidFixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
installed = load_jsonc(fixture.root_path)
|
|
write_json(
|
|
fixture.settings_path,
|
|
{"hooks": installed["hooks"], "owner": "user"},
|
|
)
|
|
paths = (*_managed_paths(fixture), fixture.settings_path)
|
|
before = snapshot(paths)
|
|
sources = read_sources()
|
|
rendered = [
|
|
render_document(document, AGENT_ID, {})
|
|
for document in source_documents(sources)
|
|
]
|
|
prepared = prepare_droid_uninstall(rendered)
|
|
real_replace = _transaction._replace_physical
|
|
failed = False
|
|
|
|
def fail_second_source(
|
|
source: str,
|
|
destination: str,
|
|
directory: _transaction.PinnedDirectory,
|
|
) -> None:
|
|
nonlocal failed
|
|
if not failed and Path(destination) == fixture.settings_path:
|
|
failed = True
|
|
raise OSError("injected Droid uninstall failure")
|
|
real_replace(source, destination, directory)
|
|
|
|
monkeypatch.setattr(_transaction, "_replace_physical", fail_second_source)
|
|
with pytest.raises(OSError, match="injected Droid uninstall failure"):
|
|
commit_droid_uninstall(prepared)
|
|
assert failed is True
|
|
assert_snapshot(before)
|
|
assert_no_transaction_files(tmp_path)
|
|
|
|
|
|
def test_cli_preflight_blocks_disabled_hooks_before_runtime_creation(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
home_dir = tmp_path / "cli-home"
|
|
settings_path = home_dir / ".factory" / "settings.json"
|
|
key_path = tmp_path / "private.key"
|
|
write_json(settings_path, {"hooksDisabled": True})
|
|
key_path.write_text(
|
|
VALID_PRIVATE_KEY,
|
|
encoding="utf-8",
|
|
)
|
|
if os.name != "nt":
|
|
key_path.chmod(0o600)
|
|
environment = {
|
|
**os.environ,
|
|
"HOME": str(home_dir),
|
|
"USERPROFILE": str(home_dir),
|
|
}
|
|
result = subprocess.run( # nosec B603
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"elydora.cli",
|
|
"install",
|
|
"--agent",
|
|
"droid",
|
|
"--org_id",
|
|
"org-1",
|
|
"--agent_id",
|
|
AGENT_ID,
|
|
"--private_key_file",
|
|
str(key_path),
|
|
"--kid",
|
|
"kid-1",
|
|
"--base_url",
|
|
"https://api.elydora.com",
|
|
],
|
|
capture_output=True,
|
|
check=False,
|
|
cwd=Path(__file__).parents[1],
|
|
env=environment,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "hooksDisabled" in result.stderr
|
|
assert not (home_dir / ".elydora").exists()
|