@@ -1135,21 +1124,21 @@
if (meta.progress_percentage !== undefined && meta.progress_percentage !== null) {
parts.push(Math.round(meta.progress_percentage) + "%");
}
- if (meta.progress_message) {
+ if (meta.progress_message && state !== "stopped") {
parts.push(meta.progress_message);
}
stateTextEl.textContent = parts.join(" · ");
}
- const canRetry = state === "failed" || !!meta.stop_requested;
- const canResume = state === "failed";
+ const canRetry = state === "failed" || state === "stopped";
+ const canResume = state === "failed" || state === "stopped";
setVisible("retry-resume-form", canRetry);
setVisible("retry-disabled", !canRetry);
setVisible("resume-disabled", !canRetry);
var resumeBtn = document.getElementById("resume-btn");
if (resumeBtn) resumeBtn.disabled = !canResume;
- const canStop = (state === "pending" || state === "processing") && !meta.stop_requested;
+ const canStop = (state === "pending" || state === "processing");
setVisible("stop-form", canStop);
setVisible("stop-disabled", !canStop);
diff --git a/frontend_multi_user/templates/plan_list.html b/frontend_multi_user/templates/plan_list.html
index e4f2e165..72f0475f 100644
--- a/frontend_multi_user/templates/plan_list.html
+++ b/frontend_multi_user/templates/plan_list.html
@@ -98,6 +98,11 @@
color: #c62828;
background: #ffebee;
}
+ .status-chip.status-stopped {
+ border-color: #e65100;
+ color: #e65100;
+ background: #fff3e0;
+ }
.plan-cell-prompt {
white-space: nowrap;
diff --git a/frontend_multi_user/templates/run_via_database.html b/frontend_multi_user/templates/run_via_database.html
index 0f900120..7604954f 100644
--- a/frontend_multi_user/templates/run_via_database.html
+++ b/frontend_multi_user/templates/run_via_database.html
@@ -60,7 +60,7 @@
-
Don't close this page. Your plan will be visible after it's generated.
+
Your plan will be visible after it's generated.
Elapsed Time: 0:00
@@ -158,20 +158,18 @@
const is_pending = data.status == "pending";
const is_processing = data.status == "processing";
const is_failed = data.status == "failed";
+ const is_stopped = data.status == "stopped";
progressMessageElement.textContent = data.progress_message || 'No progress message';
- // console.log("is_completed", is_completed);
- // console.log("is_processing", is_processing);
- // console.log("is_failed", is_failed);
if (is_pending || is_processing || is_completed) {
// There is a connection with the server, so ignore past errors
consecutive_error_count = 0;
+ } else if (is_stopped) {
+ throw new PlanExeError("Plan stopped by user");
+ } else if (is_failed) {
+ throw new PlanExeError("Failed to generate plan");
} else {
- if (is_failed) {
- throw new PlanExeError("Failed to generate plan");
- } else {
- throw new PlanExeError(`Unhandled status: ${data.status}`);
- }
+ throw new PlanExeError(`Unhandled status: ${data.status}`);
}
// Update progress bar
diff --git a/mcp_cloud/db_queries.py b/mcp_cloud/db_queries.py
index da51aca5..b0a45032 100644
--- a/mcp_cloud/db_queries.py
+++ b/mcp_cloud/db_queries.py
@@ -241,10 +241,10 @@ def _request_plan_stop_sync(plan_id: str) -> Optional[dict[str, Any]]:
if plan.state in (PlanState.pending, PlanState.processing):
plan.stop_requested = True
plan.stop_requested_timestamp = datetime.now(UTC)
- plan.state = PlanState.failed
+ plan.state = PlanState.stopped
plan.progress_message = "Stop requested by user."
db.session.commit()
- logger.info("Stop requested for plan %s; state set to failed.", plan_id)
+ logger.info("Stop requested for plan %s; state set to stopped.", plan_id)
stop_requested = True
return {
"state": get_plan_state_mapping(plan.state),
@@ -257,11 +257,11 @@ def _retry_failed_plan_sync(plan_id: str, model_profile: str, caller_metadata: O
plan = find_plan_by_id(plan_id)
if plan is None:
return None
- if plan.state != PlanState.failed:
+ if plan.state not in (PlanState.failed, PlanState.stopped):
return {
"error": {
"code": "PLAN_NOT_FAILED",
- "message": f"Plan is not in failed state: {plan_id}",
+ "message": f"Plan is not in failed or stopped state: {plan_id}",
}
}
@@ -333,11 +333,11 @@ def _resume_plan_sync(plan_id: str, model_profile: str) -> Optional[dict[str, An
plan = find_plan_by_id(plan_id)
if plan is None:
return None
- if plan.state != PlanState.failed:
+ if plan.state not in (PlanState.failed, PlanState.stopped):
return {
"error": {
"code": "PLAN_NOT_RESUMABLE",
- "message": f"Plan is not in failed state: {plan_id}",
+ "message": f"Plan is not in failed or stopped state: {plan_id}",
}
}
@@ -475,6 +475,7 @@ def get_plan_state_mapping(plan_state: PlanState) -> str:
PlanState.processing: "processing",
PlanState.completed: "completed",
PlanState.failed: "failed",
+ PlanState.stopped: "stopped",
}
return mapping.get(plan_state, "pending")
diff --git a/mcp_cloud/db_setup.py b/mcp_cloud/db_setup.py
index d2698c21..66cddc9e 100644
--- a/mcp_cloud/db_setup.py
+++ b/mcp_cloud/db_setup.py
@@ -103,10 +103,25 @@ def ensure_step_count_columns() -> None:
except Exception as exc:
logger.warning("Schema update failed for %s: %s", stmt, exc, exc_info=True)
+def ensure_stopped_state() -> None:
+ """Add 'stopped' value to the planstate/taskstate enum type (idempotent).
+
+ The PostgreSQL enum type is named ``taskstate`` in databases created before
+ the TaskState → PlanState Python rename (proposal 74). Fresh databases
+ created after that rename will have ``planstate``. We try both names.
+ """
+ with db.engine.begin() as conn:
+ for type_name in ("taskstate", "planstate"):
+ try:
+ conn.execute(text(f"ALTER TYPE {type_name} ADD VALUE IF NOT EXISTS 'stopped'"))
+ except Exception as exc:
+ logger.debug("ALTER TYPE %s: %s", type_name, exc)
+
with app.app_context():
ensure_planitem_stop_columns()
ensure_multi_api_key_columns()
ensure_step_count_columns()
+ ensure_stopped_state()
# Shown in MCP initialize (e.g. Inspector) so clients know what PlanExe does.
PLANEXE_SERVER_INSTRUCTIONS = (
@@ -136,12 +151,12 @@ def ensure_step_count_columns() -> None:
"If plan generation fails before completing all steps, call plan_resume to continue from where it left off without discarding completed work. "
"Use plan_retry instead for a full restart (plan must be in failed state). "
"Both accept a failed plan_id and optional model_profile (defaults to baseline). "
- "To stop, call plan_stop with the plan_id from plan_create; stopping is asynchronous and the plan will eventually transition to failed. "
+ "To stop, call plan_stop with the plan_id from plan_create; stopping is asynchronous and the plan will transition to the stopped state. "
"If model_profiles returns MODEL_PROFILES_UNAVAILABLE, inform the user that no models are currently configured and the server administrator needs to set up model profiles. "
"Tool errors use {error:{code,message}}. plan_file_info returns {ready:false,reason:...} while the artifact is not yet ready; check readiness by testing whether download_url is present in the response. "
"plan_file_info download_url is the absolute URL where the requested artifact can be downloaded. "
"To list recent plans for a user call plan_list; returns plan_id, state, progress_percentage, created_at, and prompt_excerpt for each plan. "
- "plan_status state contract: pending/processing => keep polling; completed => download is ready; failed => terminal error. "
+ "plan_status state contract: pending/processing => keep polling; completed => download is ready; failed => terminal error; stopped => user called plan_stop (consider plan_resume). "
"Troubleshooting: if plan_status stays in pending for longer than 5 minutes, the plan was likely queued but not picked up by a worker (server issue). "
"If plan_status is in processing and output files do not change for longer than 20 minutes, the plan_create likely failed/stalled. "
"In both cases, report the issue to PlanExe developers on GitHub: https://github.com/PlanExeOrg/PlanExe/issues . "
diff --git a/mcp_cloud/handlers.py b/mcp_cloud/handlers.py
index c95e88d6..fd2e436c 100644
--- a/mcp_cloud/handlers.py
+++ b/mcp_cloud/handlers.py
@@ -344,7 +344,7 @@ async def handle_plan_status(arguments: dict[str, Any]) -> CallToolResult:
message = plan_snapshot.get("progress_message") or "Plan generation failed."
response["error"] = {"code": "generation_failed", "message": message}
- if state not in ("completed", "failed"):
+ if state not in ("completed", "failed", "stopped"):
base_url = _get_download_base_url()
if base_url:
response["sse_url"] = f"{base_url}/sse/plan/{plan_uuid}"
@@ -575,6 +575,13 @@ async def handle_plan_file_info(arguments: dict[str, Any]) -> CallToolResult:
structuredContent=response,
isError=False,
)
+ if plan_state == PlanState.stopped:
+ response = {"ready": False, "reason": "stopped"}
+ return CallToolResult(
+ content=[TextContent(type="text", text=json.dumps(response))],
+ structuredContent=response,
+ isError=False,
+ )
if plan_state == PlanState.failed:
message = plan_snapshot["progress_message"] or "Plan generation failed."
response = {"ready": False, "reason": "failed", "error": {"code": "generation_failed", "message": message}}
diff --git a/mcp_cloud/schemas.py b/mcp_cloud/schemas.py
index 91049f61..319894da 100644
--- a/mcp_cloud/schemas.py
+++ b/mcp_cloud/schemas.py
@@ -176,7 +176,8 @@ class ToolDefinition:
"This is the primary way to check progress — it returns structured JSON with all progress fields. "
"Poll at reasonable intervals (e.g. every 5 minutes): plan generation typically takes 10-20 minutes "
"(baseline profile) and may take longer on higher-quality profiles. "
- "State contract: pending/processing => keep polling; completed => download is ready; failed => terminal error. "
+ "State contract: pending/processing => keep polling; completed => download is ready; "
+ "failed => terminal error; stopped => user called plan_stop (consider plan_resume). "
"progress_percentage is 0-100 (integer-like float); 100 when completed. "
"Note: steps vary in duration — early steps complete quickly while later steps (review, report generation) "
"take longer. Do not use progress_percentage to estimate time remaining. "
@@ -202,7 +203,7 @@ class ToolDefinition:
description=(
"Request the plan generation to stop. Pass the plan_id (the UUID returned by plan_create). "
"Stopping is asynchronous: the stop flag is set immediately but the plan may continue briefly before halting. "
- "A stopped plan will eventually transition to the failed state. "
+ "A stopped plan will transition to the stopped state. "
"If the plan is already completed or failed, stop_requested returns false (the plan already finished). "
"Unknown plan_id returns error code PLAN_NOT_FOUND."
),
@@ -218,10 +219,10 @@ class ToolDefinition:
ToolDefinition(
name="plan_retry",
description=(
- "Retry a plan that is currently in failed state. "
- "Pass the failed plan_id and optionally model_profile (defaults to baseline). "
+ "Retry a plan that is currently in failed or stopped state. "
+ "Pass the plan_id and optionally model_profile (defaults to baseline). "
"The plan is reset to pending, prior artifacts are cleared, and the same plan_id is requeued for processing. "
- "Returns PLAN_NOT_FOUND when plan_id is unknown and PLAN_NOT_FAILED when the plan is not in failed state."
+ "Returns PLAN_NOT_FOUND when plan_id is unknown and PLAN_NOT_FAILED when the plan is not in failed or stopped state."
),
input_schema=PLAN_RETRY_INPUT_SCHEMA,
output_schema=PLAN_RETRY_OUTPUT_SCHEMA,
@@ -235,13 +236,13 @@ class ToolDefinition:
ToolDefinition(
name="plan_resume",
description=(
- "Resume a failed plan without discarding completed intermediary files. "
+ "Resume a failed or stopped plan without discarding completed intermediary files. "
"Plan generation restarts from the first incomplete step, skipping all steps that already produced output files. "
- "Use plan_resume when plan_status shows 'failed' and plan generation was interrupted before completing all steps "
+ "Use plan_resume when plan_status shows 'failed' or 'stopped' and plan generation was interrupted before completing all steps "
"(network drop, timeout, plan_stop, worker crash). "
"For a full restart or to change model_profile, use plan_retry instead. "
- "Only failed plans can be resumed. "
- "Returns PLAN_NOT_FOUND when plan_id is unknown and PLAN_NOT_RESUMABLE when the plan is not in failed state. "
+ "Only failed or stopped plans can be resumed. "
+ "Returns PLAN_NOT_FOUND when plan_id is unknown and PLAN_NOT_RESUMABLE when the plan is not in failed or stopped state. "
"Returns PIPELINE_VERSION_MISMATCH when the snapshot was created by a different pipeline version; use plan_retry instead."
),
input_schema=PLAN_RESUME_INPUT_SCHEMA,
diff --git a/mcp_cloud/sse.py b/mcp_cloud/sse.py
index ff12e38f..39dbe51b 100644
--- a/mcp_cloud/sse.py
+++ b/mcp_cloud/sse.py
@@ -24,7 +24,7 @@
SSE_MAX_CONNECTIONS_PER_CLIENT: int = int(os.environ.get("PLANEXE_SSE_MAX_CONNECTIONS", "5"))
SSE_MAX_TOTAL_CONNECTIONS: int = int(os.environ.get("PLANEXE_SSE_MAX_TOTAL_CONNECTIONS", "200"))
-TERMINAL_STATES = {"completed", "failed"}
+TERMINAL_STATES = {"completed", "failed", "stopped"}
class SSEConnectionLimitError(Exception):
diff --git a/mcp_cloud/tests/test_plan_status_tool.py b/mcp_cloud/tests/test_plan_status_tool.py
index a96d6468..4c8a4d66 100644
--- a/mcp_cloud/tests/test_plan_status_tool.py
+++ b/mcp_cloud/tests/test_plan_status_tool.py
@@ -217,6 +217,49 @@ def test_plan_status_includes_current_step(self):
self.assertEqual(result.structuredContent["current_step"], "SWOT Analysis")
+ def test_plan_status_stopped_returns_stopped_state(self):
+ """User-stopped plan has state='stopped' and no stop_reason field."""
+ plan_id = str(uuid.uuid4())
+ plan_snapshot = {
+ "id": plan_id,
+ "state": PlanState.stopped,
+ "stop_requested": True,
+ "progress_percentage": 42.0,
+ "timestamp_created": datetime.now(UTC),
+ }
+ with patch(
+ "mcp_cloud.handlers._get_plan_status_snapshot_sync",
+ return_value=plan_snapshot,
+ ), patch(
+ "mcp_cloud.handlers.fetch_file_list_from_worker_plan", new=AsyncMock(return_value=[])
+ ):
+ result = asyncio.run(handle_plan_status({"plan_id": plan_id}))
+
+ self.assertEqual(result.structuredContent["state"], "stopped")
+ self.assertNotIn("stop_reason", result.structuredContent)
+
+ def test_plan_status_actual_failure_has_no_stop_reason(self):
+ """Failed plan response does not contain stop_reason field."""
+ plan_id = str(uuid.uuid4())
+ plan_snapshot = {
+ "id": plan_id,
+ "state": PlanState.failed,
+ "stop_requested": False,
+ "progress_percentage": 5.5,
+ "timestamp_created": datetime.now(UTC),
+ }
+ with patch(
+ "mcp_cloud.handlers._get_plan_status_snapshot_sync",
+ return_value=plan_snapshot,
+ ), patch(
+ "mcp_cloud.handlers.fetch_file_list_from_worker_plan", new=AsyncMock(return_value=[])
+ ):
+ result = asyncio.run(handle_plan_status({"plan_id": plan_id}))
+
+ self.assertEqual(result.structuredContent["state"], "failed")
+ self.assertNotIn("stop_reason", result.structuredContent)
+ self.assertIn("error", result.structuredContent)
+
if __name__ == "__main__":
unittest.main()
diff --git a/mcp_cloud/tool_models.py b/mcp_cloud/tool_models.py
index 7762c225..ead08213 100644
--- a/mcp_cloud/tool_models.py
+++ b/mcp_cloud/tool_models.py
@@ -107,7 +107,7 @@ class PlanStopInput(BaseModel):
class PlanRetryInput(BaseModel):
plan_id: str = Field(
...,
- description="UUID of the failed plan to retry.",
+ description="UUID of the failed or stopped plan to retry.",
)
model_profile: Literal["baseline", "premium", "frontier", "custom"] = Field(
default="baseline",
@@ -167,11 +167,12 @@ class PlanStatusSuccess(BaseModel):
...,
description="Plan UUID returned by plan_create."
)
- state: Literal["pending", "processing", "completed", "failed"] = Field(
+ state: Literal["pending", "processing", "completed", "failed", "stopped"] = Field(
...,
description=(
"Caller contract: pending/processing => keep polling; "
- "completed => download is ready; failed => terminal error."
+ "completed => download is ready; failed => terminal error; "
+ "stopped => user called plan_stop (consider plan_resume)."
),
)
progress_percentage: float = Field(
@@ -221,11 +222,12 @@ class PlanStatusOutput(BaseModel):
default=None,
description="Plan UUID returned by plan_create."
)
- state: Literal["pending", "processing", "completed", "failed"] | None = Field(
+ state: Literal["pending", "processing", "completed", "failed", "stopped"] | None = Field(
default=None,
description=(
"Caller contract: pending/processing => keep polling; "
- "completed => download is ready; failed => terminal error."
+ "completed => download is ready; failed => terminal error; "
+ "stopped => user called plan_stop (consider plan_resume)."
),
)
progress_percentage: float | None = Field(
@@ -272,7 +274,7 @@ class PlanStatusOutput(BaseModel):
class PlanStopOutput(BaseModel):
- state: Literal["pending", "processing", "completed", "failed"] | None = Field(
+ state: Literal["pending", "processing", "completed", "failed", "stopped"] | None = Field(
default=None,
description="Current plan state after stop request.",
)
@@ -286,9 +288,9 @@ class PlanStopOutput(BaseModel):
class PlanRetryOutput(BaseModel):
plan_id: str | None = Field(
default=None,
- description="Plan UUID that was retried (same ID as the failed plan).",
+ description="Plan UUID that was retried (same ID as the failed or stopped plan).",
)
- state: Literal["pending", "processing", "completed", "failed"] | None = Field(
+ state: Literal["pending", "processing", "completed", "failed", "stopped"] | None = Field(
default=None,
description="Current plan state after retry request.",
)
@@ -314,7 +316,7 @@ class PlanRetryOutput(BaseModel):
class PlanResumeInput(BaseModel):
plan_id: str = Field(
...,
- description="UUID of the failed plan to resume.",
+ description="UUID of the failed or stopped plan to resume.",
)
model_profile: Literal["baseline", "premium", "frontier", "custom"] = Field(
default="baseline",
@@ -327,9 +329,9 @@ class PlanResumeInput(BaseModel):
class PlanResumeOutput(BaseModel):
plan_id: str | None = Field(
default=None,
- description="Plan UUID that was resumed (same ID as the failed plan).",
+ description="Plan UUID that was resumed (same ID as the failed or stopped plan).",
)
- state: Literal["pending", "processing", "completed", "failed"] | None = Field(
+ state: Literal["pending", "processing", "completed", "failed", "stopped"] | None = Field(
default=None,
description="Current plan state after resume request.",
)
@@ -405,7 +407,7 @@ class PlanListInput(BaseModel):
class PlanListItem(BaseModel):
plan_id: str = Field(..., description="Plan UUID.")
- state: Literal["pending", "processing", "completed", "failed"] = Field(
+ state: Literal["pending", "processing", "completed", "failed", "stopped"] = Field(
...,
description="Current plan state.",
)
diff --git a/mcp_local/planexe_mcp_local.py b/mcp_local/planexe_mcp_local.py
index 801f92d9..facc9628 100644
--- a/mcp_local/planexe_mcp_local.py
+++ b/mcp_local/planexe_mcp_local.py
@@ -758,7 +758,8 @@ class ToolDefinition:
"Returns status and progress of the plan currently being created. "
"Poll at reasonable intervals only (e.g. every 5 minutes): plan generation typically takes 10-20 minutes "
"(baseline profile) and may take longer on higher-quality profiles. "
- "State contract: pending/processing => keep polling; completed => download is ready; failed => terminal error. "
+ "State contract: pending/processing => keep polling; completed => download is ready; "
+ "failed => terminal error; stopped => user called plan_stop (consider plan_resume). "
"progress_percentage is 0-100 (integer-like float); 100 when completed. "
"files lists intermediate outputs produced so far; use their updated_at timestamps to detect stalls. "
"Unknown plan_id returns PLAN_NOT_FOUND (or REMOTE_ERROR when transport fails). "
@@ -780,7 +781,7 @@ class ToolDefinition:
description=(
"Request the plan generation to stop. Pass the plan_id (the UUID returned by plan_create). "
"Stopping is asynchronous: the stop flag is set immediately but the plan may continue briefly before halting. "
- "A stopped plan will eventually transition to the failed state. "
+ "A stopped plan will transition to the stopped state. "
"If the plan is already completed or failed, stop_requested returns false (the plan already finished). "
"Unknown plan_id returns PLAN_NOT_FOUND (or REMOTE_ERROR when transport fails)."
),
@@ -796,10 +797,10 @@ class ToolDefinition:
ToolDefinition(
name="plan_retry",
description=(
- "Retry a plan that is currently in failed state. "
- "Pass the failed plan_id and optionally model_profile (defaults to baseline). "
+ "Retry a plan that is currently in failed or stopped state. "
+ "Pass the plan_id and optionally model_profile (defaults to baseline). "
"The same plan_id is requeued and reset to pending on the cloud service. "
- "Unknown plan_id returns PLAN_NOT_FOUND; non-failed plans return PLAN_NOT_FAILED."
+ "Unknown plan_id returns PLAN_NOT_FOUND; non-failed/stopped plans return PLAN_NOT_FAILED."
),
input_schema=PLAN_RETRY_INPUT_SCHEMA,
output_schema=PLAN_RETRY_OUTPUT_SCHEMA,
@@ -813,13 +814,13 @@ class ToolDefinition:
ToolDefinition(
name="plan_resume",
description=(
- "Resume a failed plan without discarding completed intermediary files. "
+ "Resume a failed or stopped plan without discarding completed intermediary files. "
"Plan generation restarts from the first incomplete step, skipping all steps that already produced output files. "
- "Use plan_resume when plan_status shows 'failed' and plan generation was interrupted before completing all steps "
+ "Use plan_resume when plan_status shows 'failed' or 'stopped' and plan generation was interrupted before completing all steps "
"(network drop, timeout, plan_stop, worker crash). "
"For a full restart or to change model_profile, use plan_retry instead. "
- "Only failed plans can be resumed. "
- "Returns PLAN_NOT_FOUND when plan_id is unknown and PLAN_NOT_RESUMABLE when the plan is not in failed state. "
+ "Only failed or stopped plans can be resumed. "
+ "Returns PLAN_NOT_FOUND when plan_id is unknown and PLAN_NOT_RESUMABLE when the plan is not in failed or stopped state. "
"Returns PIPELINE_VERSION_MISMATCH when the snapshot was created by a different pipeline version; use plan_retry instead."
),
input_schema=PLAN_RESUME_INPUT_SCHEMA,
@@ -894,12 +895,12 @@ class ToolDefinition:
"If plan generation fails before completing all steps, call plan_resume to continue from where it left off without discarding completed work. "
"Use plan_retry instead for a full restart. "
"Both accept the failed plan_id and optional model_profile (defaults to baseline). "
- "To stop, call plan_stop with the plan_id from plan_create; stopping is asynchronous and the plan will eventually transition to failed. "
+ "To stop, call plan_stop with the plan_id from plan_create; stopping is asynchronous and the plan will transition to the stopped state. "
"If model_profiles returns MODEL_PROFILES_UNAVAILABLE, inform the user that no models are currently configured and the server administrator needs to set up model profiles. "
"Tool errors use {error:{code,message}}. plan_download may return REMOTE_ERROR or DOWNLOAD_FAILED. "
"plan_download saves to PLANEXE_PATH (default: current working directory) and returns saved_path. "
"To list recent plans for a user call plan_list; returns plan_id, state, progress_percentage, created_at, and prompt_excerpt. "
- "plan_status state contract: pending/processing => keep polling; completed => download is ready; failed => terminal error. "
+ "plan_status state contract: pending/processing => keep polling; completed => download is ready; failed => terminal error; stopped => user called plan_stop (consider plan_resume). "
"Troubleshooting: if plan_status stays in pending for longer than 5 minutes, the plan was likely queued but not picked up by a worker (server issue). "
"If plan_status is in processing and output files do not change for longer than 20 minutes, the run likely failed/stalled. "
"In both cases, report the issue to PlanExe developers on GitHub: https://github.com/PlanExeOrg/PlanExe/issues . "
diff --git a/worker_plan_database/app.py b/worker_plan_database/app.py
index 71e35fc4..79043204 100644
--- a/worker_plan_database/app.py
+++ b/worker_plan_database/app.py
@@ -313,6 +313,20 @@ def ensure_multi_api_key_columns() -> None:
except Exception as exc:
logger.warning("Schema update failed for %s: %s", stmt, exc, exc_info=True)
+def ensure_stopped_state() -> None:
+ """Add 'stopped' value to the planstate/taskstate enum type (idempotent).
+
+ The PostgreSQL enum type is named ``taskstate`` in databases created before
+ the TaskState → PlanState Python rename (proposal 74). Fresh databases
+ created after that rename will have ``planstate``. We try both names.
+ """
+ with db.engine.begin() as conn:
+ for type_name in ("taskstate", "planstate"):
+ try:
+ conn.execute(text(f"ALTER TYPE {type_name} ADD VALUE IF NOT EXISTS 'stopped'"))
+ except Exception as exc:
+ logger.debug("ALTER TYPE %s: %s", type_name, exc)
+
def worker_process_started() -> None:
planexe_worker_id = os.environ.get("PLANEXE_WORKER_ID")
event_context = {
@@ -1040,7 +1054,8 @@ def execute_pipeline_for_job(
db.session.add(event)
db.session.commit()
else:
- update_task_state_with_retry(task_id, PlanState.failed)
+ final_state = PlanState.stopped if stop_requested else PlanState.failed
+ update_task_state_with_retry(task_id, final_state)
billing_result = _charge_usage_credits_once(task_id=task_id, run_id_dir=run_id_dir, success=False)
event_context["machai_error_message"] = machai_error_message or ""
event_context.update({
@@ -1053,7 +1068,7 @@ def execute_pipeline_for_job(
event = _new_model(
EventItem,
event_type=EventType.TASK_FAILED,
- message=f"Processing -> Failed",
+ message=f"Processing -> {'Stopped' if stop_requested else 'Failed'}",
context=event_context
)
db.session.add(event)
@@ -1314,6 +1329,7 @@ def startup_worker():
ensure_token_metrics_columns()
ensure_fractional_credit_columns()
ensure_multi_api_key_columns()
+ ensure_stopped_state()
logger.debug(f"Ensured database tables exist.")
WorkerItem.upsert_heartbeat(worker_id=WORKER_ID)
except Exception as e: