Close gaps: GLM-judge (anthropic registry) + run cancel + deep-link
ci / gates (push) Has been cancelled
ci / rust (push) Has been cancelled
ci / sandbox-k8s (push) Has been cancelled
ci / frontend (push) Has been cancelled
ci / e2e (push) Has been cancelled

#2 GLM judge (closes #114): registry NamedProvider gains a `format` field;
build_provider_registry builds an AnthropicProvider for format="anthropic".
GLM's coding/OpenAI endpoint is ToS-throttled for raw SDK, but its Anthropic
endpoint (api.z.ai/api/anthropic) accepts raw API calls (verified x-api-key
-> glm-4.7), so CLAWMATES_JUDGE_MODEL=glm:glm-4.7 routes the door governor /
topology judge through GLM with no runtime-routing. (Kimi-as-judge still needs
a Platform key — coding key is agent-only.)

#3 run-control: POST /api/topology-runs/{id}/cancel (workspace-scoped,
queued/running only); the worker honors it at the step boundary (checks
current_status in the checkpoint callback) and won't clobber a cancel with
`failed`. Frontend Run tab gains a Cancel button and clickable recent runs
that deep-link into a live/replayed stream (SSE replays from checkpoint).

cm-* tests (incl. new cancel test) + clippy + frontend lint/typecheck green.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Omar Sobh
2026-06-18 03:07:41 -07:00
co-authored by Claude Opus 4.8
parent f845dfb15f
commit 6e87433c66
10 changed files with 199 additions and 18 deletions
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT status FROM topology_runs WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "status",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "17ef083ed134d4489204d04b181ca6ed6147b2cc5bc590a3d31da60b7da77547"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE topology_runs\n SET status = 'cancelled', finished_at = now(), updated_at = now()\n WHERE id = $1 AND workspace_id = $2 AND status IN ('queued', 'running')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid"
]
},
"nullable": []
},
"hash": "88a71a747f5a87637af0b83122102a5f5e78da08268f3eb4c51a756f268f58b3"
}
+11 -3
View File
@@ -60,10 +60,18 @@ fn build_provider_registry(config: &AppConfig) -> cm_runtime::ProviderRegistry {
for p in &config.llm.providers { for p in &config.llm.providers {
match std::env::var(&p.api_key_env) { match std::env::var(&p.api_key_env) {
Ok(key) if !key.is_empty() => { Ok(key) if !key.is_empty() => {
let provider: Arc<dyn LlmProvider> = let provider: Arc<dyn LlmProvider> = match p.format.as_str() {
Arc::new(OpenAiCompatProvider::new(p.base_url.clone(), Some(key))); "anthropic" => Arc::new(cm_llm::AnthropicProvider::with_base_url(
key,
p.base_url.clone(),
)),
_ => Arc::new(OpenAiCompatProvider::new(p.base_url.clone(), Some(key))),
};
map.insert(p.name.clone(), provider); map.insert(p.name.clone(), provider);
println!("clawmates-server: registered LLM provider '{}'", p.name); println!(
"clawmates-server: registered LLM provider '{}' ({})",
p.name, p.format
);
} }
_ => eprintln!( _ => eprintln!(
"clawmates-server: provider '{}' skipped — {} is unset", "clawmates-server: provider '{}' skipped — {} is unset",
+4
View File
@@ -150,6 +150,10 @@ pub fn router(state: AppState) -> Router {
"/api/topology-runs/{id}/events", "/api/topology-runs/{id}/events",
get(routes::topology::run_events_sse), get(routes::topology::run_events_sse),
) )
.route(
"/api/topology-runs/{id}/cancel",
post(routes::topology::cancel_run),
)
.layer(tower_http::trace::TraceLayer::new_for_http()) .layer(tower_http::trace::TraceLayer::new_for_http())
.with_state(state) .with_state(state)
} }
+16
View File
@@ -275,6 +275,22 @@ pub async fn run_events_sse(
Sse::new(stream).keep_alive(KeepAlive::default()) Sse::new(stream).keep_alive(KeepAlive::default())
} }
/// `POST /api/topology-runs/{id}/cancel` — request cancellation of a queued or
/// running job; the worker stops at its next step boundary. 409 if the run is
/// already terminal or unknown.
pub async fn cancel_run(
State(state): State<AppState>,
Authed(user): Authed,
Path(id): Path<Uuid>,
) -> Result<StatusCode, ApiError> {
let cancelled = cm_db::repo::topology_runs::cancel(&state.pool, id, user.workspace_id).await?;
if cancelled {
Ok(StatusCode::OK)
} else {
Err(ApiError::Conflict)
}
}
/// `GET /api/topology-runs/{id}` — a single run with status + result. /// `GET /api/topology-runs/{id}` — a single run with status + result.
pub async fn get_run( pub async fn get_run(
State(state): State<AppState>, State(state): State<AppState>,
+16
View File
@@ -81,6 +81,15 @@ async fn run_job(pool: &PgPool, job: cm_db::repo::topology_runs::ClaimedTopology
cm_db::repo::topology_runs::checkpoint(&pool, id, &v, snap.completed as i64) cm_db::repo::topology_runs::checkpoint(&pool, id, &v, snap.completed as i64)
.await; .await;
} }
// Honor cancellation at the step boundary: stop before the next turn.
if matches!(
cm_db::repo::topology_runs::current_status(&pool, id).await,
Ok(Some(ref s)) if s == "cancelled"
) {
return Err(cm_orchestrator::OrchestratorError::Executor(
"run cancelled".into(),
));
}
Ok(()) Ok(())
} }
}) })
@@ -94,7 +103,14 @@ async fn run_job(pool: &PgPool, job: cm_db::repo::topology_runs::ClaimedTopology
} }
} }
Err(e) => { Err(e) => {
// Don't clobber a cancellation (or any already-terminal state) with `failed`.
let terminal = matches!(
cm_db::repo::topology_runs::current_status(pool, id).await,
Ok(Some(ref s)) if s == "cancelled" || s == "completed" || s == "failed"
);
if !terminal {
let _ = cm_db::repo::topology_runs::fail(pool, id, &format!("{e}")).await; let _ = cm_db::repo::topology_runs::fail(pool, id, &format!("{e}")).await;
} }
} }
} }
}
+27
View File
@@ -85,3 +85,30 @@ async fn stale_running_jobs_are_requeued_for_resume() {
assert_eq!(st.status, "queued"); assert_eq!(st.status, "queued");
assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_some()); assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_some());
} }
#[tokio::test]
async fn cancel_transitions_only_active_runs() {
let pool = cm_testkit::test_pool().await;
let ws = seed_workspace(&pool).await;
let id = Uuid::now_v7();
let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
.await
.unwrap();
// A queued run cancels; the worker sees the new status (no workspace scope).
assert!(topology_runs::cancel(&pool, id, ws).await.unwrap());
assert_eq!(
topology_runs::current_status(&pool, id).await.unwrap().as_deref(),
Some("cancelled")
);
assert_eq!(topology_runs::status(&pool, id, ws).await.unwrap().status, "cancelled");
// Already terminal → cannot cancel again; wrong workspace → no-op.
assert!(!topology_runs::cancel(&pool, id, ws).await.unwrap());
let other = seed_workspace(&pool).await;
let id2 = Uuid::now_v7();
topology_runs::enqueue_run(&pool, id2, ws, "t", &graph).await.unwrap();
assert!(!topology_runs::cancel(&pool, id2, other).await.unwrap());
}
+14 -4
View File
@@ -69,17 +69,27 @@ pub struct LlmConfig {
pub providers: Vec<NamedProvider>, pub providers: Vec<NamedProvider>,
} }
/// An extra OpenAI-compatible provider in the registry (e.g. GLM or Kimi). /// An extra provider in the registry (e.g. GLM or Kimi). Used as judge or
/// topology-exec model via `"<name>:<model>"`.
#[derive(Debug, Clone, Deserialize)] #[derive(Debug, Clone, Deserialize)]
pub struct NamedProvider { pub struct NamedProvider {
/// Selector prefix, e.g. `glm` or `kimi` (used as `"glm:glm-4.6"`). /// Selector prefix, e.g. `glm` or `kimi` (used as `"glm:glm-4.6"`).
pub name: String, pub name: String,
/// OpenAI-compatible base URL incl. version, e.g. /// Base URL incl. version, e.g. `https://open.bigmodel.cn/api/paas/v4`
/// `https://open.bigmodel.cn/api/paas/v4` (GLM) or /// (OpenAI-compat) or `https://api.z.ai/api/anthropic` (Anthropic-format).
/// `https://api.moonshot.ai/v1` (Kimi).
pub base_url: String, pub base_url: String,
/// Env var holding this provider's API key, e.g. `GLM_API_KEY`. /// Env var holding this provider's API key, e.g. `GLM_API_KEY`.
pub api_key_env: String, pub api_key_env: String,
/// Wire format: `openai_compat` (default) or `anthropic`. GLM's coding
/// OpenAI endpoint is ToS-throttled for raw SDK access, but its Anthropic
/// endpoint (`api.z.ai/api/anthropic`) accepts raw API calls — so a GLM
/// judge uses `format = "anthropic"`.
#[serde(default = "default_provider_format")]
pub format: String,
}
fn default_provider_format() -> String {
"openai_compat".to_string()
} }
#[derive(Debug, Clone, Deserialize)] #[derive(Debug, Clone, Deserialize)]
+25
View File
@@ -176,6 +176,31 @@ pub async fn fail(pool: &PgPool, id: Uuid, error: &str) -> Result<(), DbError> {
Ok(()) Ok(())
} }
/// Cancel a run (workspace-scoped). Only `queued`/`running` jobs can be
/// cancelled; returns whether a row transitioned. The worker observes the new
/// status at its next step boundary and stops.
pub async fn cancel(pool: &PgPool, id: Uuid, workspace_id: WorkspaceId) -> Result<bool, DbError> {
let result = sqlx::query!(
"UPDATE topology_runs
SET status = 'cancelled', finished_at = now(), updated_at = now()
WHERE id = $1 AND workspace_id = $2 AND status IN ('queued', 'running')",
id,
workspace_id.as_uuid(),
)
.execute(pool)
.await?;
Ok(result.rows_affected() == 1)
}
/// The current status of a run (no workspace scope) — used by the worker to
/// detect cancellation mid-run without re-reading the whole row.
pub async fn current_status(pool: &PgPool, id: Uuid) -> Result<Option<String>, DbError> {
let row = sqlx::query!("SELECT status FROM topology_runs WHERE id = $1", id)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| r.status))
}
/// Resume sweep: requeue `running` jobs whose worker went silent (no checkpoint /// Resume sweep: requeue `running` jobs whose worker went silent (no checkpoint
/// touch within `older_than_secs`). The next claim resumes them from checkpoint. /// touch within `older_than_secs`). The next claim resumes them from checkpoint.
/// Returns how many were requeued. /// Returns how many were requeued.
@@ -43,6 +43,7 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
const [busy, setBusy] = useState(false); const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null); const [error, setError] = useState<string | null>(null);
const [runs, setRuns] = useState<RunSummary[]>([]); const [runs, setRuns] = useState<RunSummary[]>([]);
const [runId, setRunId] = useState<string | null>(null);
const esRef = useRef<EventSource | null>(null); const esRef = useRef<EventSource | null>(null);
async function loadRuns() { async function loadRuns() {
@@ -68,6 +69,7 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
* browser auto-sends Last-Event-ID on reconnect so the server resumes. */ * browser auto-sends Last-Event-ID on reconnect so the server resumes. */
function stream(id: string) { function stream(id: string) {
esRef.current?.close(); esRef.current?.close();
setRunId(id);
const es = new EventSource(`/api/topology-runs/${id}/events`); const es = new EventSource(`/api/topology-runs/${id}/events`);
esRef.current = es; esRef.current = es;
setStatus("running"); setStatus("running");
@@ -139,6 +141,27 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
} }
} }
/** Request cancellation; the SSE `done` event reports the cancelled status. */
async function cancel() {
if (!runId) return;
try {
await fetch(`/api/topology-runs/${runId}/cancel`, { method: "POST" });
} catch {
/* the worker also stops on the next status read */
}
}
/** Deep-link into a past/running run: the SSE endpoint replays its steps from
* the checkpoint, then tails live if it's still running. */
function openRun(id: string) {
setError(null);
setSteps([]);
setFinalOutput(null);
setStatus("running");
setBusy(true);
stream(id);
}
const started = status !== null; const started = status !== null;
return ( return (
@@ -175,7 +198,7 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
/> />
</label> </label>
</div> </div>
<div> <div className="flex gap-2">
<button <button
type="button" type="button"
onClick={run} onClick={run}
@@ -184,6 +207,15 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
> >
{busy ? "Running…" : "Run topology"} {busy ? "Running…" : "Run topology"}
</button> </button>
{busy ? (
<button
type="button"
onClick={cancel}
className="rounded-lg border border-border px-4 py-2 text-sm text-muted-foreground hover:text-foreground"
>
Cancel
</button>
) : null}
</div> </div>
{error ? <p className="text-sm text-red-500">{error}</p> : null} {error ? <p className="text-sm text-red-500">{error}</p> : null}
@@ -230,7 +262,12 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
<p className="mb-2 text-xs font-medium text-muted-foreground">Recent runs</p> <p className="mb-2 text-xs font-medium text-muted-foreground">Recent runs</p>
<ul className="flex flex-col gap-1"> <ul className="flex flex-col gap-1">
{runs.slice(0, 8).map((r) => ( {runs.slice(0, 8).map((r) => (
<li key={r.id} className="flex items-center gap-2 text-xs text-muted-foreground"> <li key={r.id}>
<button
type="button"
onClick={() => openRun(r.id)}
className="flex w-full items-center gap-2 rounded-md px-1 py-1 text-left text-xs text-muted-foreground hover:bg-muted/40"
>
<span <span
className={`rounded-full px-2 py-0.5 ${ className={`rounded-full px-2 py-0.5 ${
STATUS_STYLE[r.status] ?? "bg-muted text-muted-foreground" STATUS_STYLE[r.status] ?? "bg-muted text-muted-foreground"
@@ -240,6 +277,7 @@ export function TopologyRun({ catalog }: { catalog: CatalogEntry[] }) {
</span> </span>
<span className="text-foreground">{r.kind}</span> <span className="text-foreground">{r.kind}</span>
<span className="truncate">{r.task}</span> <span className="truncate">{r.task}</span>
</button>
</li> </li>
))} ))}
</ul> </ul>