diff --git a/.sqlx/query-6f55f9867b0fdaeedaa5995462e70e3b2118fb595c8d6e313b5f0271c0f5d537.json b/.sqlx/query-30960f21f8c28a4190efacb6a68d5918da107b23a74b34d5dff0ee34e90a51b2.json similarity index 58% rename from .sqlx/query-6f55f9867b0fdaeedaa5995462e70e3b2118fb595c8d6e313b5f0271c0f5d537.json rename to .sqlx/query-30960f21f8c28a4190efacb6a68d5918da107b23a74b34d5dff0ee34e90a51b2.json index 19beba0..4d4e703 100644 --- a/.sqlx/query-6f55f9867b0fdaeedaa5995462e70e3b2118fb595c8d6e313b5f0271c0f5d537.json +++ b/.sqlx/query-30960f21f8c28a4190efacb6a68d5918da107b23a74b34d5dff0ee34e90a51b2.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n status,\n progress_json,\n result_json,\n error_json,\n to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,\n to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",\n to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"updated_at!\",\n to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at\n from async_jobs\n where workspace_id = $1\n and ($2::text is null or agent_id = $2)\n and ($3::text is null or operation_id = $3)\n and ($4::text is null or status = $4)\n order by updated_at desc\n limit $5", + "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n status,\n progress_json,\n result_json,\n error_json,\n expires_at as \"expires_at: OffsetDateTime\",\n created_at as \"created_at!: OffsetDateTime\",\n updated_at as \"updated_at!: OffsetDateTime\",\n finished_at as \"finished_at: OffsetDateTime\"\n from async_jobs\n where workspace_id = $1\n and ($2::text is null or agent_id = $2)\n and ($3::text is null or operation_id = $3)\n and ($4::text is null or status = $4)\n order by updated_at desc\n limit $5", "describe": { "columns": [ { @@ -45,23 +45,23 @@ }, { "ordinal": 8, - "name": "expires_at", - "type_info": "Text" + "name": "expires_at: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 9, - "name": "created_at!", - "type_info": "Text" + "name": "created_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 10, - "name": "updated_at!", - "type_info": "Text" + "name": "updated_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 11, - "name": "finished_at", - "type_info": "Text" + "name": "finished_at: OffsetDateTime", + "type_info": "Timestamptz" } ], "parameters": { @@ -82,11 +82,11 @@ false, true, true, - null, - null, - null, - null + true, + false, + false, + true ] }, - "hash": "6f55f9867b0fdaeedaa5995462e70e3b2118fb595c8d6e313b5f0271c0f5d537" + "hash": "30960f21f8c28a4190efacb6a68d5918da107b23a74b34d5dff0ee34e90a51b2" } diff --git a/.sqlx/query-e92340e11c1b1d5dcd1441f12588e01fa18acc43e18316ea9789cd60e4967eaf.json b/.sqlx/query-6ea069ced9ab34bbd92d2068e457d17c0f98483284e55548c6f1f24ac284ae11.json similarity index 63% rename from .sqlx/query-e92340e11c1b1d5dcd1441f12588e01fa18acc43e18316ea9789cd60e4967eaf.json rename to .sqlx/query-6ea069ced9ab34bbd92d2068e457d17c0f98483284e55548c6f1f24ac284ae11.json index 49f0f2a..d5ce7bb 100644 --- a/.sqlx/query-e92340e11c1b1d5dcd1441f12588e01fa18acc43e18316ea9789cd60e4967eaf.json +++ b/.sqlx/query-6ea069ced9ab34bbd92d2068e457d17c0f98483284e55548c6f1f24ac284ae11.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n status,\n progress_json,\n result_json,\n error_json,\n to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,\n to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",\n to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"updated_at!\",\n to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at\n from async_jobs\n where id = $1", + "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n status,\n progress_json,\n result_json,\n error_json,\n expires_at as \"expires_at: OffsetDateTime\",\n created_at as \"created_at!: OffsetDateTime\",\n updated_at as \"updated_at!: OffsetDateTime\",\n finished_at as \"finished_at: OffsetDateTime\"\n from async_jobs\n where id = $1", "describe": { "columns": [ { @@ -45,23 +45,23 @@ }, { "ordinal": 8, - "name": "expires_at", - "type_info": "Text" + "name": "expires_at: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 9, - "name": "created_at!", - "type_info": "Text" + "name": "created_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 10, - "name": "updated_at!", - "type_info": "Text" + "name": "updated_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 11, - "name": "finished_at", - "type_info": "Text" + "name": "finished_at: OffsetDateTime", + "type_info": "Timestamptz" } ], "parameters": { @@ -78,11 +78,11 @@ false, true, true, - null, - null, - null, - null + true, + false, + false, + true ] }, - "hash": "e92340e11c1b1d5dcd1441f12588e01fa18acc43e18316ea9789cd60e4967eaf" + "hash": "6ea069ced9ab34bbd92d2068e457d17c0f98483284e55548c6f1f24ac284ae11" } diff --git a/.sqlx/query-0c383ab1d4c5c178f9ded8219a81da145a4ba6198a67de5c7e060803619ad7fc.json b/.sqlx/query-a1d2c9b16b61d226701449fc22146db6f0f7d17e6867f929c44027990fb94b57.json similarity index 64% rename from .sqlx/query-0c383ab1d4c5c178f9ded8219a81da145a4ba6198a67de5c7e060803619ad7fc.json rename to .sqlx/query-a1d2c9b16b61d226701449fc22146db6f0f7d17e6867f929c44027990fb94b57.json index d0f7393..8900459 100644 --- a/.sqlx/query-0c383ab1d4c5c178f9ded8219a81da145a4ba6198a67de5c7e060803619ad7fc.json +++ b/.sqlx/query-a1d2c9b16b61d226701449fc22146db6f0f7d17e6867f929c44027990fb94b57.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n protocol,\n mode,\n status,\n cursor_json,\n state_json,\n to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"expires_at!\",\n to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at,\n to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",\n to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at\n from stream_sessions\n where id = $1", + "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n protocol,\n mode,\n status,\n cursor_json,\n state_json,\n expires_at as \"expires_at!: OffsetDateTime\",\n last_poll_at as \"last_poll_at: OffsetDateTime\",\n created_at as \"created_at!: OffsetDateTime\",\n closed_at as \"closed_at: OffsetDateTime\"\n from stream_sessions\n where id = $1", "describe": { "columns": [ { @@ -50,23 +50,23 @@ }, { "ordinal": 9, - "name": "expires_at!", - "type_info": "Text" + "name": "expires_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 10, - "name": "last_poll_at", - "type_info": "Text" + "name": "last_poll_at: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 11, - "name": "created_at!", - "type_info": "Text" + "name": "created_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 12, - "name": "closed_at", - "type_info": "Text" + "name": "closed_at: OffsetDateTime", + "type_info": "Timestamptz" } ], "parameters": { @@ -84,11 +84,11 @@ false, true, false, - null, - null, - null, - null + false, + true, + false, + true ] }, - "hash": "0c383ab1d4c5c178f9ded8219a81da145a4ba6198a67de5c7e060803619ad7fc" + "hash": "a1d2c9b16b61d226701449fc22146db6f0f7d17e6867f929c44027990fb94b57" } diff --git a/.sqlx/query-4e2091cb75e67ac455b0efee795e94b8a18807a59a0f11796fec27519f34817d.json b/.sqlx/query-b68377d8ffd5bdc9b8e417170f4940b65d2167dada6b6110985ce0bd95480b93.json similarity index 58% rename from .sqlx/query-4e2091cb75e67ac455b0efee795e94b8a18807a59a0f11796fec27519f34817d.json rename to .sqlx/query-b68377d8ffd5bdc9b8e417170f4940b65d2167dada6b6110985ce0bd95480b93.json index 6a3e3c5..21fc532 100644 --- a/.sqlx/query-4e2091cb75e67ac455b0efee795e94b8a18807a59a0f11796fec27519f34817d.json +++ b/.sqlx/query-b68377d8ffd5bdc9b8e417170f4940b65d2167dada6b6110985ce0bd95480b93.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n protocol,\n mode,\n status,\n cursor_json,\n state_json,\n to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"expires_at!\",\n to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at,\n to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",\n to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at\n from stream_sessions\n where workspace_id = $1\n and ($2::text is null or agent_id = $2)\n and ($3::text is null or operation_id = $3)\n and ($4::text is null or status = $4)\n and ($5::text is null or mode = $5)\n order by created_at desc\n limit $6", + "query": "select\n id,\n workspace_id,\n agent_id,\n operation_id,\n protocol,\n mode,\n status,\n cursor_json,\n state_json,\n expires_at as \"expires_at!: OffsetDateTime\",\n last_poll_at as \"last_poll_at: OffsetDateTime\",\n created_at as \"created_at!: OffsetDateTime\",\n closed_at as \"closed_at: OffsetDateTime\"\n from stream_sessions\n where workspace_id = $1\n and ($2::text is null or agent_id = $2)\n and ($3::text is null or operation_id = $3)\n and ($4::text is null or status = $4)\n and ($5::text is null or mode = $5)\n order by created_at desc\n limit $6", "describe": { "columns": [ { @@ -50,23 +50,23 @@ }, { "ordinal": 9, - "name": "expires_at!", - "type_info": "Text" + "name": "expires_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 10, - "name": "last_poll_at", - "type_info": "Text" + "name": "last_poll_at: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 11, - "name": "created_at!", - "type_info": "Text" + "name": "created_at!: OffsetDateTime", + "type_info": "Timestamptz" }, { "ordinal": 12, - "name": "closed_at", - "type_info": "Text" + "name": "closed_at: OffsetDateTime", + "type_info": "Timestamptz" } ], "parameters": { @@ -89,11 +89,11 @@ false, true, false, - null, - null, - null, - null + false, + true, + false, + true ] }, - "hash": "4e2091cb75e67ac455b0efee795e94b8a18807a59a0f11796fec27519f34817d" + "hash": "b68377d8ffd5bdc9b8e417170f4940b65d2167dada6b6110985ce0bd95480b93" } diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 2eb8ad2..02a4680 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -153,7 +153,8 @@ pub struct WindowTestRunView { pub struct StreamSessionStartView { pub session_id: String, pub status: StreamStatus, - pub expires_at: String, + #[serde(with = "time::serde::rfc3339")] + pub expires_at: OffsetDateTime, pub poll_after_ms: u64, pub preview: Value, } @@ -225,7 +226,7 @@ impl TestRunOutcome { json!({ "session_id": output.session_id, "status": output.status, - "expires_at": output.expires_at, + "expires_at": format_timestamp(output.expires_at), "poll_after_ms": output.poll_after_ms, "preview": output.preview, }), @@ -451,10 +452,14 @@ pub struct StreamSessionSummaryView { pub protocol: Protocol, pub mode: ExecutionMode, pub status: StreamStatus, - pub expires_at: String, - pub last_poll_at: Option, - pub created_at: String, - pub closed_at: Option, + #[serde(with = "time::serde::rfc3339")] + pub expires_at: OffsetDateTime, + #[serde(with = "time::serde::rfc3339::option")] + pub last_poll_at: Option, + #[serde(with = "time::serde::rfc3339")] + pub created_at: OffsetDateTime, + #[serde(with = "time::serde::rfc3339::option")] + pub closed_at: Option, } #[derive(Clone, Debug, Serialize)] @@ -480,10 +485,14 @@ pub struct AsyncJobSummaryView { pub agent_id: Option, pub status: JobStatus, pub progress: Value, - pub created_at: String, - pub updated_at: String, - pub finished_at: Option, - pub expires_at: Option, + #[serde(with = "time::serde::rfc3339")] + pub created_at: OffsetDateTime, + #[serde(with = "time::serde::rfc3339")] + pub updated_at: OffsetDateTime, + #[serde(with = "time::serde::rfc3339::option")] + pub finished_at: Option, + #[serde(with = "time::serde::rfc3339::option")] + pub expires_at: Option, } #[derive(Clone, Debug, Serialize)] @@ -1551,7 +1560,7 @@ impl AdminService { })?; ensure_stream_session_workspace(&session, workspace_id)?; self.registry - .close_stream_session(session_id, &now_string()?) + .close_stream_session(session_id, &OffsetDateTime::now_utc()) .await?; let updated = self .registry @@ -1621,7 +1630,7 @@ impl AdminService { })?; ensure_async_job_workspace(&job, workspace_id)?; self.registry - .cancel_async_job(job_id, &now_string()?) + .cancel_async_job(job_id, &OffsetDateTime::now_utc()) .await?; let updated = self.registry.get_async_job(job_id).await?.ok_or_else(|| { ApiError::not_found(format!("async job {} was not found", job_id.as_str())) @@ -2186,16 +2195,14 @@ impl AdminService { let preview_count = seed.items.len().min(batch_size); let preview_items = seed.items[..preview_count].to_vec(); let next_index = preview_count; - let created_at = now_string().map_err(|error| RuntimeError::SecretCrypto { - details: error.to_string(), - })?; - let expires_at = add_millis( - &created_at, - streaming.max_session_lifetime_ms.unwrap_or(60_000), - ) - .map_err(|error| RuntimeError::SecretCrypto { - details: error.to_string(), - })?; + let created_at = OffsetDateTime::now_utc(); + let expires_at = created_at + .checked_add(time::Duration::milliseconds( + streaming.max_session_lifetime_ms.unwrap_or(60_000) as i64, + )) + .ok_or_else(|| RuntimeError::SecretCrypto { + details: "failed to compute stream session expiration".to_owned(), + })?; let session = StreamSession { id: crank_core::StreamSessionId::new(new_prefixed_id("sess")), workspace_id: workspace_id.clone(), @@ -2212,8 +2219,8 @@ impl AdminService { next_index, batch_size, }), - expires_at: expires_at.clone(), - last_poll_at: Some(created_at.clone()), + expires_at, + last_poll_at: Some(created_at), created_at, closed_at: None, }; @@ -2243,21 +2250,19 @@ impl AdminService { runtime: &RuntimeOperation, input: &Value, ) -> Result { - let created_at = now_string().map_err(|error| RuntimeError::SecretCrypto { - details: error.to_string(), - })?; + let created_at = OffsetDateTime::now_utc(); let streaming = runtime.execution_config.streaming.as_ref().ok_or_else(|| { RuntimeError::MissingStreamingConfig { operation_id: runtime.operation_id.as_str().to_owned(), } })?; - let expires_at = add_millis( - &created_at, - streaming.max_session_lifetime_ms.unwrap_or(300_000), - ) - .map_err(|error| RuntimeError::SecretCrypto { - details: error.to_string(), - })?; + let expires_at = created_at + .checked_add(time::Duration::milliseconds( + streaming.max_session_lifetime_ms.unwrap_or(300_000) as i64, + )) + .ok_or_else(|| RuntimeError::SecretCrypto { + details: "failed to compute async job expiration".to_owned(), + })?; let job = AsyncJobHandle { id: crank_core::AsyncJobId::new(new_prefixed_id("job")), workspace_id: workspace_id.clone(), @@ -2268,8 +2273,8 @@ impl AdminService { result: None, error: None, expires_at: Some(expires_at), - created_at: created_at.clone(), - updated_at: created_at.clone(), + created_at, + updated_at: created_at, finished_at: None, }; self.registry @@ -2306,10 +2311,7 @@ impl AdminService { } Err(error) => Err(error), }; - let finished_at = match now_string() { - Ok(value) => value, - Err(_) => return, - }; + let finished_at = OffsetDateTime::now_utc(); let _ = match result { Ok(output) => { registry @@ -4613,14 +4615,10 @@ fn now_string() -> Result { .map_err(|error| ApiError::internal(error.to_string())) } -fn add_millis(timestamp: &str, millis: u64) -> Result { - let parsed = OffsetDateTime::parse(timestamp, &Rfc3339) - .map_err(|error| ApiError::internal(error.to_string()))?; - parsed - .checked_add(time::Duration::milliseconds(millis as i64)) - .ok_or_else(|| ApiError::internal("failed to add millis to timestamp"))? +fn format_timestamp(timestamp: OffsetDateTime) -> String { + timestamp .format(&Rfc3339) - .map_err(|error| ApiError::internal(error.to_string())) + .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_owned()) } async fn resolve_runtime_auth_for_task( diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index 752e92e..0cddd9c 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -741,8 +741,8 @@ async fn handle_session_start_call( let next_index = preview_count; let session_id = StreamSessionId::new(format!("sess_{}", uuid::Uuid::now_v7().simple())); - let now = now_rfc3339(); - let expires_at = add_millis(&now, streaming.max_session_lifetime_ms.unwrap_or(60_000)); + let now = OffsetDateTime::now_utc(); + let expires_at = add_millis(now, streaming.max_session_lifetime_ms.unwrap_or(60_000)); let session_record = StreamSession { id: session_id.clone(), workspace_id: tool.workspace_id.clone(), @@ -759,9 +759,9 @@ async fn handle_session_start_call( next_index, batch_size, }), - expires_at: expires_at.clone(), - last_poll_at: Some(now.clone()), - created_at: now.clone(), + expires_at, + last_poll_at: Some(now), + created_at: now, closed_at: None, }; @@ -778,7 +778,7 @@ async fn handle_session_start_call( let output = json!({ "session_id": session_id.as_str(), "status": "running", - "expires_at": expires_at, + "expires_at": format_rfc3339(expires_at), "poll_after_ms": streaming.poll_interval_ms.unwrap_or(1000), "preview": { "summary": seed.summary, @@ -873,7 +873,7 @@ async fn handle_session_poll_call( ); }; - let now = now_rfc3339(); + let now = OffsetDateTime::now_utc(); let loaded = match state .registry .get_stream_session(&StreamSessionId::new(control.session_id.clone())) @@ -902,7 +902,7 @@ async fn handle_session_poll_call( ); } - if loaded.is_expired(&now) { + if loaded.is_expired(now) { let _ = state .registry .update_stream_session_state(UpdateStreamSessionStateRequest { @@ -926,7 +926,7 @@ async fn handle_session_poll_call( ); } - if !loaded.can_poll(&now) { + if !loaded.can_poll(now) { return tool_error_response( message, response_mode, @@ -970,11 +970,11 @@ async fn handle_session_poll_call( cursor: cursor.as_ref(), state: &json!(state_payload), expires_at: Some(&add_millis( - &now, + now, streaming.max_session_lifetime_ms.unwrap_or(60_000), )), last_poll_at: Some(&now), - closed_at: matches!(next_status, StreamStatus::Stopped).then_some(now.as_str()), + closed_at: matches!(next_status, StreamStatus::Stopped).then_some(&now), }) .await { @@ -985,7 +985,7 @@ async fn handle_session_poll_call( let output = json!({ "session_id": updated.id.as_str(), "status": serialize_stream_status(updated.status), - "expires_at": updated.expires_at, + "expires_at": format_rfc3339(updated.expires_at), "summary": serde_json::from_value::(updated.state.clone()).map(|value| value.summary).unwrap_or(Value::Null), "items": items, "cursor": updated.cursor, @@ -1069,7 +1069,7 @@ async fn handle_session_stop_call( .registry .close_stream_session( &StreamSessionId::new(control.session_id.clone()), - &now_rfc3339(), + &OffsetDateTime::now_utc(), ) .await { @@ -1106,7 +1106,7 @@ async fn handle_async_job_start_call( ); }; - let now = now_rfc3339(); + let now = OffsetDateTime::now_utc(); let job = AsyncJobHandle { id: AsyncJobId::new(format!("job_{}", uuid::Uuid::now_v7().simple())), workspace_id: tool.workspace_id.clone(), @@ -1117,11 +1117,11 @@ async fn handle_async_job_start_call( result: None, error: None, expires_at: Some(add_millis( - &now, + now, streaming.max_session_lifetime_ms.unwrap_or(300_000), )), - created_at: now.clone(), - updated_at: now.clone(), + created_at: now, + updated_at: now, finished_at: None, }; @@ -1156,7 +1156,7 @@ async fn handle_async_job_start_call( } Err(error) => Err(error), }; - let finished_at = now_rfc3339(); + let finished_at = OffsetDateTime::now_utc(); let update_result = match result { Ok(output) => { @@ -1424,7 +1424,10 @@ async fn handle_async_job_cancel_call( match state .registry - .cancel_async_job(&AsyncJobId::new(control.job_id.clone()), &now_rfc3339()) + .cancel_async_job( + &AsyncJobId::new(control.job_id.clone()), + &OffsetDateTime::now_utc(), + ) .await { Ok(()) => success_tool_response( @@ -1909,21 +1912,20 @@ fn tool_error_response( ) } -fn now_rfc3339() -> String { - OffsetDateTime::now_utc() +fn format_rfc3339(timestamp: OffsetDateTime) -> String { + timestamp .format(&Rfc3339) .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_owned()) } -fn add_millis(timestamp: &str, millis: u64) -> String { - let Ok(parsed) = OffsetDateTime::parse(timestamp, &Rfc3339) else { - return timestamp.to_owned(); - }; +fn now_rfc3339() -> String { + format_rfc3339(OffsetDateTime::now_utc()) +} + +fn add_millis(timestamp: OffsetDateTime, millis: u64) -> OffsetDateTime { let delta = time::Duration::milliseconds(i64::try_from(millis).unwrap_or(i64::MAX)); - (parsed + delta) - .format(&Rfc3339) - .unwrap_or_else(|_| timestamp.to_owned()) + timestamp + delta } fn serialize_stream_status(status: StreamStatus) -> &'static str { diff --git a/crates/crank-core/src/stream_session.rs b/crates/crank-core/src/stream_session.rs index b510936..70c90d6 100644 --- a/crates/crank-core/src/stream_session.rs +++ b/crates/crank-core/src/stream_session.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; +use time::OffsetDateTime; use crate::{ ids::{AgentId, AsyncJobId, OperationId, StreamSessionId, WorkspaceId}, @@ -36,31 +37,34 @@ pub struct StreamSession { #[serde(skip_serializing_if = "Option::is_none")] pub cursor: Option, pub state: Value, - pub expires_at: String, + #[serde(with = "time::serde::rfc3339")] + pub expires_at: OffsetDateTime, #[serde(skip_serializing_if = "Option::is_none")] - pub last_poll_at: Option, - pub created_at: String, + #[serde(with = "time::serde::rfc3339::option")] + pub last_poll_at: Option, + #[serde(with = "time::serde::rfc3339")] + pub created_at: OffsetDateTime, #[serde(skip_serializing_if = "Option::is_none")] - pub closed_at: Option, + #[serde(with = "time::serde::rfc3339::option")] + pub closed_at: Option, } impl StreamSession { - pub fn is_expired(&self, now: &str) -> bool { - self.expires_at.as_str() <= now + pub fn is_expired(&self, now: OffsetDateTime) -> bool { + self.expires_at <= now } - pub fn can_poll(&self, now: &str) -> bool { + pub fn can_poll(&self, now: OffsetDateTime) -> bool { !self.status.is_terminal() && !self.is_expired(now) } - pub fn mark_polled(&mut self, now: impl Into) { - self.last_poll_at = Some(now.into()); + pub fn mark_polled(&mut self, now: OffsetDateTime) { + self.last_poll_at = Some(now); } - pub fn mark_closed(&mut self, now: impl Into) { - let now = now.into(); + pub fn mark_closed(&mut self, now: OffsetDateTime) { self.status = StreamStatus::Stopped; - self.last_poll_at = Some(now.clone()); + self.last_poll_at = Some(now); self.closed_at = Some(now); } } @@ -99,11 +103,15 @@ pub struct AsyncJobHandle { #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub expires_at: Option, - pub created_at: String, - pub updated_at: String, + #[serde(with = "time::serde::rfc3339::option")] + pub expires_at: Option, + #[serde(with = "time::serde::rfc3339")] + pub created_at: OffsetDateTime, + #[serde(with = "time::serde::rfc3339")] + pub updated_at: OffsetDateTime, #[serde(skip_serializing_if = "Option::is_none")] - pub finished_at: Option, + #[serde(with = "time::serde::rfc3339::option")] + pub finished_at: Option, } impl AsyncJobHandle { @@ -115,20 +123,18 @@ impl AsyncJobHandle { matches!(self.status, JobStatus::Created | JobStatus::Running) } - pub fn mark_finished(&mut self, now: impl Into, result: Value) { - let now = now.into(); + pub fn mark_finished(&mut self, now: OffsetDateTime, result: Value) { self.status = JobStatus::Completed; self.result = Some(result); self.error = None; - self.updated_at = now.clone(); + self.updated_at = now; self.finished_at = Some(now); } - pub fn mark_failed(&mut self, now: impl Into, error: Value) { - let now = now.into(); + pub fn mark_failed(&mut self, now: OffsetDateTime, error: Value) { self.status = JobStatus::Failed; self.error = Some(error); - self.updated_at = now.clone(); + self.updated_at = now; self.finished_at = Some(now); } } @@ -136,6 +142,7 @@ impl AsyncJobHandle { #[cfg(test)] mod tests { use serde_json::json; + use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use super::{AsyncJobHandle, JobStatus, StreamSession, StreamStatus}; use crate::{ @@ -144,6 +151,10 @@ mod tests { streaming::ExecutionMode, }; + fn timestamp(value: &str) -> OffsetDateTime { + OffsetDateTime::parse(value, &Rfc3339).unwrap() + } + #[test] fn stream_session_tracks_poll_and_close_transitions() { let mut session = StreamSession { @@ -156,20 +167,20 @@ mod tests { status: StreamStatus::Running, cursor: None, state: json!({"cursor":"abc"}), - expires_at: "2026-04-06T12:05:00Z".to_owned(), + expires_at: timestamp("2026-04-06T12:05:00Z"), last_poll_at: None, - created_at: "2026-04-06T12:00:00Z".to_owned(), + created_at: timestamp("2026-04-06T12:00:00Z"), closed_at: None, }; - assert!(session.can_poll("2026-04-06T12:01:00Z")); + assert!(session.can_poll(timestamp("2026-04-06T12:01:00Z"))); - session.mark_polled("2026-04-06T12:01:00Z"); - session.mark_closed("2026-04-06T12:02:00Z"); + session.mark_polled(timestamp("2026-04-06T12:01:00Z")); + session.mark_closed(timestamp("2026-04-06T12:02:00Z")); assert_eq!(session.status, StreamStatus::Stopped); - assert_eq!(session.closed_at.as_deref(), Some("2026-04-06T12:02:00Z")); - assert!(!session.can_poll("2026-04-06T12:03:00Z")); + assert_eq!(session.closed_at, Some(timestamp("2026-04-06T12:02:00Z"))); + assert!(!session.can_poll(timestamp("2026-04-06T12:03:00Z"))); } #[test] @@ -183,15 +194,15 @@ mod tests { progress: json!({"percent": 60}), result: None, error: None, - expires_at: Some("2026-04-06T12:05:00Z".to_owned()), - created_at: "2026-04-06T12:00:00Z".to_owned(), - updated_at: "2026-04-06T12:00:00Z".to_owned(), + expires_at: Some(timestamp("2026-04-06T12:05:00Z")), + created_at: timestamp("2026-04-06T12:00:00Z"), + updated_at: timestamp("2026-04-06T12:00:00Z"), finished_at: None, }; assert!(job.can_cancel()); - job.mark_finished("2026-04-06T12:01:00Z", json!({"ok": true})); + job.mark_finished(timestamp("2026-04-06T12:01:00Z"), json!({"ok": true})); assert!(job.is_finished()); assert_eq!(job.status, JobStatus::Completed); @@ -200,11 +211,14 @@ mod tests { failed_job.result = None; failed_job.finished_at = None; - failed_job.mark_failed("2026-04-06T12:02:00Z", json!({"message": "boom"})); + failed_job.mark_failed( + timestamp("2026-04-06T12:02:00Z"), + json!({"message": "boom"}), + ); assert_eq!(failed_job.status, JobStatus::Failed); assert_eq!( - failed_job.finished_at.as_deref(), - Some("2026-04-06T12:02:00Z") + failed_job.finished_at, + Some(timestamp("2026-04-06T12:02:00Z")) ); } } diff --git a/crates/crank-registry/src/model.rs b/crates/crank-registry/src/model.rs index ca2155c..26264fd 100644 --- a/crates/crank-registry/src/model.rs +++ b/crates/crank-registry/src/model.rs @@ -10,6 +10,7 @@ use crank_mapping::MappingSet; use crank_schema::Schema; use serde::{Deserialize, Serialize}; use serde_json::Value; +use time::OffsetDateTime; macro_rules! define_registry_id { ($name:ident) => { @@ -464,9 +465,9 @@ pub struct UpdateStreamSessionStateRequest<'a> { pub next_status: StreamStatus, pub cursor: Option<&'a Value>, pub state: &'a Value, - pub expires_at: Option<&'a str>, - pub last_poll_at: Option<&'a str>, - pub closed_at: Option<&'a str>, + pub expires_at: Option<&'a OffsetDateTime>, + pub last_poll_at: Option<&'a OffsetDateTime>, + pub closed_at: Option<&'a OffsetDateTime>, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -492,9 +493,9 @@ pub struct UpdateAsyncJobStatusRequest<'a> { pub progress: &'a Value, pub result: Option<&'a Value>, pub error: Option<&'a Value>, - pub expires_at: Option<&'a str>, - pub updated_at: &'a str, - pub finished_at: Option<&'a str>, + pub expires_at: Option<&'a OffsetDateTime>, + pub updated_at: &'a OffsetDateTime, + pub finished_at: Option<&'a OffsetDateTime>, } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index bed5922..ae79bf3 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -2015,7 +2015,7 @@ mod tests { .await .unwrap(); registry - .close_stream_session(&closable_session.id, "2026-04-06T12:00:30Z") + .close_stream_session(&closable_session.id, ×tamp("2026-04-06T12:00:30Z")) .await .unwrap(); @@ -2033,14 +2033,14 @@ mod tests { next_status: StreamStatus::Failed, cursor: Some(&json!({"cursor":"next"})), state: &json!({"phase":"failed"}), - expires_at: Some("2026-04-06T12:10:00Z"), - last_poll_at: Some("2026-04-06T12:01:00Z"), - closed_at: Some("2026-04-06T12:01:00Z"), + expires_at: Some(×tamp("2026-04-06T12:10:00Z")), + last_poll_at: Some(×tamp("2026-04-06T12:01:00Z")), + closed_at: Some(×tamp("2026-04-06T12:01:00Z")), }) .await .unwrap(); assert_eq!(updated.status, StreamStatus::Failed); - assert_eq!(updated.closed_at.as_deref(), Some("2026-04-06T12:01:00Z")); + assert_eq!(updated.closed_at, Some(timestamp("2026-04-06T12:01:00Z"))); let page = registry .list_stream_sessions(StreamSessionFilter { @@ -2062,7 +2062,7 @@ mod tests { .create_stream_session(CreateStreamSessionRequest { session: &StreamSession { id: expired_session.id.clone(), - expires_at: "2026-04-06T11:00:00Z".to_owned(), + expires_at: timestamp("2026-04-06T11:00:00Z"), ..expired_session }, }) @@ -2070,7 +2070,7 @@ mod tests { .unwrap(); let deleted = registry - .delete_expired_stream_sessions("2026-04-06T12:00:00Z") + .delete_expired_stream_sessions(×tamp("2026-04-06T12:00:00Z")) .await .unwrap(); assert_eq!(deleted, 1); @@ -2140,7 +2140,7 @@ mod tests { .await .unwrap(); registry - .cancel_async_job(&cancellable_job.id, "2026-04-06T12:00:30Z") + .cancel_async_job(&cancellable_job.id, ×tamp("2026-04-06T12:00:30Z")) .await .unwrap(); @@ -2159,9 +2159,9 @@ mod tests { progress: &json!({"percent":100}), result: Some(&json!({"ok":true})), error: None, - expires_at: Some("2026-04-06T12:15:00Z"), - updated_at: "2026-04-06T12:01:00Z", - finished_at: Some("2026-04-06T12:01:00Z"), + expires_at: Some(×tamp("2026-04-06T12:15:00Z")), + updated_at: ×tamp("2026-04-06T12:01:00Z"), + finished_at: Some(×tamp("2026-04-06T12:01:00Z")), }) .await .unwrap(); @@ -2185,7 +2185,7 @@ mod tests { let expired_job = AsyncJobHandle { id: crank_core::AsyncJobId::new("job_expired"), - expires_at: Some("2026-04-06T11:00:00Z".to_owned()), + expires_at: Some(timestamp("2026-04-06T11:00:00Z")), ..test_async_job("job_expired", "op_job_01", JobStatus::Cancelled) }; registry @@ -2194,7 +2194,7 @@ mod tests { .unwrap(); let deleted = registry - .delete_expired_async_jobs("2026-04-06T12:00:00Z") + .delete_expired_async_jobs(×tamp("2026-04-06T12:00:00Z")) .await .unwrap(); assert_eq!(deleted, 1); @@ -2227,7 +2227,7 @@ mod tests { result: None, error: None, expires_at: None, - updated_at: "2026-04-06T12:01:00Z", + updated_at: ×tamp("2026-04-06T12:01:00Z"), finished_at: None, }) .await @@ -2432,9 +2432,9 @@ mod tests { status, cursor: Some(json!({"cursor":"initial"})), state: json!({"phase":"running"}), - expires_at: "2026-04-06T12:05:00Z".to_owned(), + expires_at: timestamp("2026-04-06T12:05:00Z"), last_poll_at: None, - created_at: "2026-04-06T12:00:00Z".to_owned(), + created_at: timestamp("2026-04-06T12:00:00Z"), closed_at: None, } } @@ -2510,9 +2510,9 @@ mod tests { progress: json!({"percent": 25}), result: None, error: None, - expires_at: Some("2026-04-06T12:05:00Z".to_owned()), - created_at: "2026-04-06T12:00:00Z".to_owned(), - updated_at: "2026-04-06T12:00:00Z".to_owned(), + expires_at: Some(timestamp("2026-04-06T12:05:00Z")), + created_at: timestamp("2026-04-06T12:00:00Z"), + updated_at: timestamp("2026-04-06T12:00:00Z"), finished_at: None, } } diff --git a/crates/crank-registry/src/postgres/stream.rs b/crates/crank-registry/src/postgres/stream.rs index 05e66c7..a036af4 100644 --- a/crates/crank-registry/src/postgres/stream.rs +++ b/crates/crank-registry/src/postgres/stream.rs @@ -1,4 +1,5 @@ use super::*; +use time::OffsetDateTime; impl PostgresRegistry { pub async fn create_stream_session( @@ -40,10 +41,10 @@ impl PostgresRegistry { .bind(serialize_enum_text(&request.session.status, "status")?) .bind(request.session.cursor.clone().map(Json)) .bind(Json(request.session.state.clone())) - .bind(&request.session.expires_at) - .bind(request.session.last_poll_at.as_deref()) - .bind(&request.session.created_at) - .bind(request.session.closed_at.as_deref()) + .bind(request.session.expires_at) + .bind(request.session.last_poll_at) + .bind(request.session.created_at) + .bind(request.session.closed_at) .execute(&self.pool) .await?; @@ -65,10 +66,10 @@ impl PostgresRegistry { status, cursor_json, state_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"expires_at!\", - to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\", - to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at + expires_at as \"expires_at!: OffsetDateTime\", + last_poll_at as \"last_poll_at: OffsetDateTime\", + created_at as \"created_at!: OffsetDateTime\", + closed_at as \"closed_at: OffsetDateTime\" from stream_sessions where id = $1", id.as_str(), @@ -129,10 +130,10 @@ impl PostgresRegistry { status, cursor_json, state_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at", + expires_at, + last_poll_at, + created_at, + closed_at", ) .bind(request.session_id.as_str()) .bind(serialize_enum_text(&request.current_status, "status")?) @@ -166,7 +167,7 @@ impl PostgresRegistry { pub async fn close_stream_session( &self, id: &StreamSessionId, - now: &str, + now: &OffsetDateTime, ) -> Result<(), RegistryError> { let result = sqlx::query( "update stream_sessions @@ -224,10 +225,10 @@ impl PostgresRegistry { status, cursor_json, state_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"expires_at!\", - to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\", - to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at + expires_at as \"expires_at!: OffsetDateTime\", + last_poll_at as \"last_poll_at: OffsetDateTime\", + created_at as \"created_at!: OffsetDateTime\", + closed_at as \"closed_at: OffsetDateTime\" from stream_sessions where workspace_id = $1 and ($2::text is null or agent_id = $2) @@ -271,7 +272,10 @@ impl PostgresRegistry { Ok(Page { items, total }) } - pub async fn delete_expired_stream_sessions(&self, now: &str) -> Result { + pub async fn delete_expired_stream_sessions( + &self, + now: &OffsetDateTime, + ) -> Result { let result = sqlx::query( "delete from stream_sessions where expires_at <= $1::timestamptz", @@ -314,10 +318,10 @@ impl PostgresRegistry { .bind(Json(request.job.progress.clone())) .bind(request.job.result.clone().map(Json)) .bind(request.job.error.clone().map(Json)) - .bind(request.job.expires_at.as_deref()) - .bind(&request.job.created_at) - .bind(&request.job.updated_at) - .bind(request.job.finished_at.as_deref()) + .bind(request.job.expires_at) + .bind(request.job.created_at) + .bind(request.job.updated_at) + .bind(request.job.finished_at) .execute(&self.pool) .await?; @@ -338,10 +342,10 @@ impl PostgresRegistry { progress_json, result_json, error_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\", - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"updated_at!\", - to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at + expires_at as \"expires_at: OffsetDateTime\", + created_at as \"created_at!: OffsetDateTime\", + updated_at as \"updated_at!: OffsetDateTime\", + finished_at as \"finished_at: OffsetDateTime\" from async_jobs where id = $1", id.as_str(), @@ -397,10 +401,10 @@ impl PostgresRegistry { progress_json, result_json, error_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at", + expires_at, + created_at, + updated_at, + finished_at", ) .bind(request.job_id.as_str()) .bind(serialize_enum_text(&request.current_status, "status")?) @@ -432,7 +436,11 @@ impl PostgresRegistry { } } - pub async fn cancel_async_job(&self, id: &AsyncJobId, now: &str) -> Result<(), RegistryError> { + pub async fn cancel_async_job( + &self, + id: &AsyncJobId, + now: &OffsetDateTime, + ) -> Result<(), RegistryError> { let result = sqlx::query( "update async_jobs set status = 'cancelled', @@ -483,10 +491,10 @@ impl PostgresRegistry { progress_json, result_json, error_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\", - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"updated_at!\", - to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at + expires_at as \"expires_at: OffsetDateTime\", + created_at as \"created_at!: OffsetDateTime\", + updated_at as \"updated_at!: OffsetDateTime\", + finished_at as \"finished_at: OffsetDateTime\" from async_jobs where workspace_id = $1 and ($2::text is null or agent_id = $2) @@ -527,7 +535,10 @@ impl PostgresRegistry { Ok(Page { items, total }) } - pub async fn delete_expired_async_jobs(&self, now: &str) -> Result { + pub async fn delete_expired_async_jobs( + &self, + now: &OffsetDateTime, + ) -> Result { let result = sqlx::query( "delete from async_jobs where expires_at is not null