mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-12 04:37:44 +02:00
[eric] workflows: a deleted workflow stays deleted, so the immortal 45-minute ghost can't write itself back to life
This commit is contained in:
@@ -213,9 +213,13 @@ async def execute(
|
||||
# of a toggled-off workflow running itself. Turn it back on to run it.
|
||||
p_live = storage.get_workflow(wf.id)
|
||||
p_refusal = None
|
||||
if p_live is not None and p_live.deleted_at is not None:
|
||||
if p_live is None:
|
||||
# A hard delete leaves nothing to look up, and reading that as "no objection" is how a
|
||||
# deleted workflow still ran to the end and then wrote itself back to life.
|
||||
p_refusal = "Workflow deleted"
|
||||
elif p_live is not None and not p_live.schedule.enabled:
|
||||
elif p_live.deleted_at is not None:
|
||||
p_refusal = "Workflow deleted"
|
||||
elif not p_live.schedule.enabled:
|
||||
p_refusal = "Workflow is paused"
|
||||
if p_refusal is not None:
|
||||
p_skipped = WorkflowRun(
|
||||
|
||||
@@ -36,6 +36,13 @@ _missed_cache: list[MissedRun] = []
|
||||
_cache_loaded = False
|
||||
_paused = False
|
||||
|
||||
# Ids deleted during this process's life. The cache hands out SHARED Workflow instances, so a run
|
||||
# already in flight when the user deletes still holds one and writes it back when it finishes, and
|
||||
# save_workflow used to recreate the file AND the cache entry, fully scheduled: the workflow rose
|
||||
# from the dead every time, which is exactly the "it never dies" field report. Only needs to live in
|
||||
# memory, because after a restart nothing holds a stale instance to write back.
|
||||
p_deleted_ids: set[str] = set()
|
||||
|
||||
|
||||
def _resolve_host_tz_name() -> str:
|
||||
"""Best-effort IANA name for the host. Mirrors apps/service/client.py."""
|
||||
@@ -158,8 +165,18 @@ def get_workflow(wid: str) -> Optional[Workflow]:
|
||||
return _workflow_cache.get(wid)
|
||||
|
||||
|
||||
def save_workflow(wf: Workflow) -> Workflow:
|
||||
def save_workflow(wf: Workflow, untrash: bool = False) -> Workflow:
|
||||
with _io_lock:
|
||||
if wf.id in p_deleted_ids:
|
||||
logger.info("ignoring a write-back for deleted workflow %s", wf.id)
|
||||
return wf
|
||||
# Trash is one-way too: only /restore passes untrash. Any other save carrying an older copy
|
||||
# (a run that started before the user hit delete) would otherwise clear deleted_at and put
|
||||
# the workflow back on the page with its schedule re-armed.
|
||||
prior = _workflow_cache.get(wf.id)
|
||||
if not untrash and prior is not None and prior.deleted_at is not None and wf.deleted_at is None:
|
||||
logger.info("ignoring a write-back that would untrash workflow %s", wf.id)
|
||||
return wf
|
||||
_ensure_dirs()
|
||||
_workflow_cache[wf.id] = wf
|
||||
p_atomic_write_json(_wf_path(wf.id), wf.model_dump(mode="json"))
|
||||
@@ -171,6 +188,9 @@ def reload_workflow(wid: str) -> Optional[Workflow]:
|
||||
instances, so a handler that mutated one and then failed must roll back through here or the
|
||||
unsaved change lingers until any later save persists it by accident."""
|
||||
with _io_lock:
|
||||
if wid in p_deleted_ids:
|
||||
_workflow_cache.pop(wid, None)
|
||||
return None
|
||||
path = _wf_path(wid)
|
||||
if not os.path.exists(path):
|
||||
_workflow_cache.pop(wid, None)
|
||||
@@ -184,6 +204,7 @@ def reload_workflow(wid: str) -> Optional[Workflow]:
|
||||
def delete_workflow(wid: str) -> bool:
|
||||
with _io_lock:
|
||||
existed = wid in _workflow_cache
|
||||
p_deleted_ids.add(wid)
|
||||
_workflow_cache.pop(wid, None)
|
||||
_runs_cache.pop(wid, None)
|
||||
wf_file = _wf_path(wid)
|
||||
@@ -217,6 +238,8 @@ def list_all_runs(limit: int = 200) -> list[WorkflowRun]:
|
||||
|
||||
def record_run(run: WorkflowRun) -> WorkflowRun:
|
||||
with _io_lock:
|
||||
if run.workflow_id in p_deleted_ids:
|
||||
return run
|
||||
_ensure_dirs()
|
||||
arr = _runs_cache.setdefault(run.workflow_id, [])
|
||||
# Replace prior entry with same id if we're updating an in-flight run.
|
||||
|
||||
@@ -906,7 +906,7 @@ async def restore_workflow(workflow_id: str):
|
||||
if not wf or wf.deleted_at is None:
|
||||
raise HTTPException(status_code=404, detail="Workflow not in trash")
|
||||
wf.deleted_at = None
|
||||
storage.save_workflow(wf)
|
||||
storage.save_workflow(wf, untrash=True)
|
||||
enriched = _enriched(wf)
|
||||
try:
|
||||
from backend.apps.agents.core.ws_manager import ws_manager
|
||||
@@ -1453,6 +1453,10 @@ async def run_workflow_now(workflow_id: str, body: Optional[dict] = None):
|
||||
wf = storage.get_workflow(workflow_id)
|
||||
if not wf:
|
||||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||||
# The executor refuses a trashed workflow but writes no history for it, so without this the caller
|
||||
# got run_id "" with a null status and no idea why nothing happened.
|
||||
if wf.deleted_at is not None:
|
||||
raise HTTPException(status_code=409, detail="This workflow is in Trash. Restore it to run it.")
|
||||
# executor.execute() owns the run record. Don't pre-create a stub here or we end up with two rows per manual fire (one orphan "running" row from this handler plus the real one from the executor).
|
||||
pre_ids = {r.id for r in storage.list_runs(wf.id, limit=10)}
|
||||
tested_signature = body.get("signature") if isinstance(body, dict) else None
|
||||
|
||||
Reference in New Issue
Block a user