fix(workflow): restore registry entry verbatim on post-removal rmtree failure

workflow_remove now persists registry.remove() before deleting any
files (fixed previously), but if the registry write succeeds and the
subsequent shutil.rmtree(workflow_dir) then fails, the registry was
left claiming the workflow uninstalled while its directory remained on
disk -- an orphaned install with no path back to a clean state.
workflow_step_remove already handles this exact sequencing by capturing
the registry entry before removal and restoring it directly into
registry.data plus save() (bypassing add(), which would stamp a new
updated_at) if the directory removal fails afterwards.

Applied the same pattern to workflow_remove: capture registry_metadata
via registry.get() before registry.remove(), and on an rmtree OSError,
write it straight back into registry.data["workflows"][workflow_id] and
save(), matching workflow_step_remove's restore-failure handling (a
yellow warning, not a hard failure, since the primary error is already
about to be reported). Existing error message and exit behavior for the
rmtree failure are unchanged.

Added a failing-first regression: install a workflow, monkeypatch
shutil.rmtree to raise OSError, and assert a clean existing error
message, the directory remaining (rmtree never actually deleted
anything), and the registry entry restored byte-for-byte identical
(including installed_at/updated_at) -- proving the fix bypasses add()
and doesn't re-stamp timestamps. Confirmed red (registry entry stayed
None) before the fix, green after.

Assisted-by: GitHub Copilot (model: Claude Sonnet 5, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-11 08:55:31 +02:00
parent 49f6fb6279
commit 812050a4f9
2 changed files with 59 additions and 0 deletions

View File

@@ -1122,6 +1122,11 @@ def workflow_remove(
)
raise typer.Exit(1)
# Captured before the registry write so a subsequent directory-removal
# failure can restore it verbatim (bypassing add(), which would stamp a
# new updated_at), mirroring workflow_step_remove's same restore pattern.
registry_metadata = registry.get(workflow_id)
# Persist the registry removal before touching any files: if save()
# fails, WorkflowRegistry.remove() rolls back its in-memory state and
# raises, so the workflow stays fully installed (files + registry) rather
@@ -1140,6 +1145,18 @@ def workflow_remove(
try:
shutil.rmtree(workflow_dir)
except OSError as exc:
# The registry removal already succeeded; restore the original
# entry verbatim so the registry doesn't claim this workflow is
# uninstalled while its directory is still sitting on disk.
try:
if registry_metadata is not None:
registry.data["workflows"][workflow_id] = registry_metadata
registry.save()
except Exception as restore_exc: # noqa: BLE001
console.print(
f"[yellow]Warning:[/yellow] Failed to restore registry entry "
f"for '{safe_id}' after directory removal failure: {restore_exc}"
)
console.print(
f"[red]Error:[/red] Failed to remove workflow directory {workflow_dir}: {exc}"
)

View File

@@ -5935,6 +5935,48 @@ class TestWorkflowRemoveGuard:
# The on-disk registry must still claim the workflow installed.
assert WorkflowRegistry(project_dir).is_installed("test-wf")
def test_remove_directory_failure_restores_registry_entry_verbatim(
self, project_dir, monkeypatch
):
"""If the registry removal already persisted successfully but the
subsequent shutil.rmtree fails, the directory was never actually
deleted (rmtree raised before removing anything usable), so the
registry must not be left claiming the workflow uninstalled. The
restored entry must be byte-for-byte the original (same
installed_at/updated_at) -- calling add() again would stamp a new
updated_at, which is why workflow_step_remove restores directly via
registry.data and save() instead of add()."""
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry
registry = WorkflowRegistry(project_dir)
registry.add("test-wf", {"name": "Test", "version": "1.0.0"})
workflow_dir = project_dir / ".specify" / "workflows" / "test-wf"
workflow_dir.mkdir(parents=True, exist_ok=True)
(workflow_dir / "workflow.yml").write_text("keep-me", encoding="utf-8")
original_entry = WorkflowRegistry(project_dir).get("test-wf")
def boom(*args, **kwargs):
raise OSError("permission denied")
monkeypatch.chdir(project_dir)
with pytest.MonkeyPatch.context() as mp:
mp.setattr("shutil.rmtree", boom)
result = CliRunner().invoke(app, ["workflow", "remove", "test-wf"])
assert result.exit_code != 0
assert result.exception is None or isinstance(result.exception, SystemExit)
assert "Failed to remove workflow directory" in result.output
# rmtree raised, so nothing was actually deleted.
assert workflow_dir.exists()
assert (workflow_dir / "workflow.yml").read_text(encoding="utf-8") == "keep-me"
# The registry entry must come back exactly as it was, not re-added.
restored = WorkflowRegistry(project_dir).get("test-wf")
assert restored == original_entry
assert WorkflowRegistry(project_dir).is_installed("test-wf")
class TestWorkflowAddSymlinkGuard:
def test_add_malformed_ipv6_url_exits_cleanly(self, temp_dir, monkeypatch):