simplify(automation): hold one lifetime file lease per task key

crates/tracedecay-automation-runtime/src/automation/effect_runtime/settlement/tests.rs

@@ -389,19 +389,10 @@ async fn retained_disabled_user_job_run(
     .await
 }
 
-fn now_secs() -> i64 {
-    std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .expect("wall clock")
-        .as_secs() as i64
-}
-
 async fn task_lock_is_denied(dashboard_root: &std::path::Path, job_id: &str) -> bool {
     crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed(
         dashboard_root,
         &format!("user_job_{job_id}"),
-        None,
-        now_secs(),
     )
     .await
     .expect("competing task lock")
@@ -2237,8 +2228,6 @@ async fn retained_projector_panic_finishes_recovery_before_releasing_task_lock()
         if crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed(
             dashboard_root,
             &format!("user_job_{job_id}"),
-            None,
-            now_secs(),
         )
         .await
         .expect("post-projector-panic task lock")
@@ -2581,8 +2570,6 @@ async fn retained_pair_attempts_second_leg_and_keeps_both_guards_until_both_fini
         let reacquired = crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed(
             dashboard_root,
             &format!("user_job_{job_id}"),
-            None,
-            now_secs(),
         )
         .await
         .expect("post-pair lock probe");
@@ -3010,8 +2997,6 @@ async fn retained_user_job_rebinds_and_recovery_retires_only_terminal_corrupt_sp
         crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed(
             dashboard_root,
             &format!("user_job_{job_id}"),
-            None,
-            now_secs(),
         )
         .await
         .expect("post-terminal lock")

crates/tracedecay-automation-runtime/src/automation/jobs.rs

@@ -44,7 +44,6 @@ const JOB_COMMAND_TIMEOUT_SECS: u64 = 30;
 const JOB_COMMAND_OUTPUT_CAP_CHARS: usize = 16 * 1024;
 const JOB_SKILL_BODY_CAP_CHARS: usize = 4_000;
 const DEFAULT_JOB_FAILURE_COOLDOWN_SECS: u64 = 300;
-const DEFAULT_JOB_STALE_LOCK_SECS: u64 = 6 * 60 * 60;
 const WEBHOOK_TIMEOUT_SECS: u64 = 10;
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
@@ -179,15 +178,8 @@ fn job_lock_key(job_id: &str) -> String {
 pub(super) async fn try_acquire_job_task_lock(
     dashboard_root: &Path,
     job_id: &str,
-    now_secs: i64,
 ) -> Result<Option<AutomationTaskLock>> {
-    AutomationTaskLock::try_acquire_keyed(
-        dashboard_root,
-        &job_lock_key(job_id),
-        Some(DEFAULT_JOB_STALE_LOCK_SECS),
-        now_secs,
-    )
-    .await
+    AutomationTaskLock::try_acquire_keyed(dashboard_root, &job_lock_key(job_id)).await
 }
 
 #[tracing::instrument(name = "automation.jobs.load", level = "trace", skip_all)]
@@ -546,9 +538,7 @@ async fn run_retained_user_job(
         return ctx.skipped(reason).await.map_err(Into::into);
     }
 
-    let now_secs = current_timestamp();
-    let Some(task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id, now_secs).await?
-    else {
+    let Some(task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id).await? else {
         let reason = if trigger == AutomationTrigger::Scheduler {
             AutomationSkipReasonV1::SchedulerLockActive
         } else {

crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_gate.rs

@@ -57,9 +57,7 @@ pub async fn evaluate_and_record_scheduler_skip(
         .map(Some);
     }
 
-    let lock_time = current_timestamp();
-    let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id, lock_time).await?
-    else {
+    let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id).await? else {
         crate::automation::scheduler_trace::observe_skip_reason(
             AutomationSkipReasonV1::SchedulerLockActive,
         );
@@ -123,8 +121,7 @@ pub(super) async fn evaluate_and_record_scheduler_skip_at(
         .map(Some);
     }
 
-    let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id, now_secs).await?
-    else {
+    let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id).await? else {
         return record_scheduler_lock_skip(
             dashboard_root,
             config,

crates/tracedecay-automation-runtime/src/automation/lifecycle.rs

@@ -26,8 +26,7 @@ use super::run_ledger::{
     AutomationTrigger, append_run_record, load_run_ledger_task_summary,
 };
 use super::scheduler::{
-    AutomationTaskLock, load_session_activity, schedule_decision, stale_lock_secs,
-    task_disabled_skip,
+    AutomationTaskLock, load_session_activity, schedule_decision, task_disabled_skip,
 };
 use tracedecay_domain::errors::{Result, TraceDecayError};
 use tracedecay_global_db::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1};
@@ -529,16 +528,7 @@ fn scheduler_gate_with_lock_retention<'a>(
                 trigger,
                 AutomationTrigger::Scheduler | AutomationTrigger::HostReceipt
             );
-
-            let lock_now_secs = current_timestamp();
-            let Some(lock) = AutomationTaskLock::try_acquire(
-                dashboard_root,
-                task,
-                stale_lock_secs(config, task),
-                lock_now_secs,
-            )
-            .await?
-            else {
+            let Some(lock) = AutomationTaskLock::try_acquire(dashboard_root, task).await? else {
                 super::scheduler_trace::observe_skip_reason(
                     AutomationSkipReasonV1::SchedulerLockActive,
                 );
@@ -1434,15 +1424,11 @@ mod recorded_failure_tests {
         let root = tempfile::tempdir().expect("temporary dashboard root");
         let settlement_guard = AutomationRunSettlementGuard::new();
         let retention = settlement_guard.retention();
-        let task_lock = AutomationTaskLock::try_acquire(
-            root.path(),
-            AgentTaskKind::SessionReflector,
-            Some(3_600),
-            current_timestamp(),
-        )
-        .await
-        .expect("first lock acquisition")
-        .expect("first lock");
+        let task_lock =
+            AutomationTaskLock::try_acquire(root.path(), AgentTaskKind::SessionReflector)
+                .await
+                .expect("first lock acquisition")
+                .expect("first lock");
         assert!(
             retain_task_lock(task_lock, Some(&retention))
                 .expect("retain lock")
@@ -1459,28 +1445,34 @@ mod recorded_failure_tests {
         assert!(matches!(result, Err(AutomationRunError::Runtime(_))));
 
         assert!(
-            AutomationTaskLock::try_acquire(
-                root.path(),
-                AgentTaskKind::SessionReflector,
-                Some(3_600),
-                current_timestamp(),
-            )
-            .await
-            .expect("competing lock acquisition")
-            .is_none()
+            AutomationTaskLock::try_acquire(root.path(), AgentTaskKind::SessionReflector,)
+                .await
+                .expect("competing lock acquisition")
+                .is_none()
         );
 
+        let old_coordination = std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(
+                root.path()
+                    .join("automation_locks/session_reflector.lock.lock"),
+            )
+            .expect("old publisher coordination");
+        assert!(matches!(
+            old_coordination.try_lock(),
+            Err(std::fs::TryLockError::WouldBlock)
+        ));
         drop(settlement_guard);
+        old_coordination
+            .try_lock()
+            .expect("settlement releases old publisher coordination");
+        old_coordination.unlock().expect("release old publisher");
         assert!(
-            AutomationTaskLock::try_acquire(
-                root.path(),
-                AgentTaskKind::SessionReflector,
-                Some(3_600),
-                current_timestamp(),
-            )
-            .await
-            .expect("post-settlement lock acquisition")
-            .is_some()
+            AutomationTaskLock::try_acquire(root.path(), AgentTaskKind::SessionReflector,)
+                .await
+                .expect("post-settlement lock acquisition")
+                .is_some()
         );
     }
 }