fix(diagnostics): resolve superseded job failures

crates/cli/src/commands/doctor.rs

@@ -7,7 +7,7 @@ use platform::{
     TrustDomainState,
 };
 use serde::Serialize;
-use state::{Database, JobStatus, PvPaths, RuntimeObservedStatus, StateError};
+use state::{Database, JobDiagnosticSubject, PvPaths, RuntimeObservedStatus, StateError};
 
 use crate::args::DoctorArgs;
 use crate::environment::Environment;
@@ -138,7 +138,7 @@ struct DoctorCheck {
     name: &'static str,
     message: String,
     detail: Option<String>,
-    repair: Option<&'static str>,
+    repair: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     routing: Option<PfRoutingDiagnostic>,
 }
@@ -161,18 +161,18 @@ impl DoctorCheck {
             name,
             message: message.into(),
             detail: None,
-            repair,
+            repair: repair.map(str::to_owned),
             routing: None,
         }
     }
 
-    fn fail(name: &'static str, message: impl Into<String>, repair: &'static str) -> Self {
+    fn fail(name: &'static str, message: impl Into<String>, repair: impl Into<String>) -> Self {
         Self {
             status: CheckStatus::Fail,
             name,
             message: message.into(),
             detail: None,
-            repair: Some(repair),
+            repair: Some(repair.into()),
             routing: None,
         }
     }
@@ -482,33 +482,32 @@ fn recent_jobs_check(database: Option<&Database>) -> Result<DoctorCheck, Execute
             Some("pv setup"),
         ));
     };
-    let failed = database
-        .recent_jobs()?
-        .into_iter()
-        .filter(|job| job.status == JobStatus::Failed)
-        .collect::<Vec<_>>();
+    let failed = database.unresolved_job_failures()?;
 
     if failed.is_empty() {
         return Ok(DoctorCheck::pass(
             "Recent jobs",
-            "no failed jobs in recent history",
+            "no unresolved failed jobs",
         ));
     }
 
+    let repair = repair_for_job_subject(&failed[0].subject);
     Ok(DoctorCheck::fail(
         "Recent jobs",
-        format!("{} failed job(s) in recent history", failed.len()),
-        "pv setup",
+        format!("{} unresolved failed job(s)", failed.len()),
+        repair,
     )
     .with_detail(
         failed
             .into_iter()
-            .map(|job| {
+            .map(|failure| {
+                let job = failure.job;
                 format!(
-                    "{} {} {}: {}",
+                    "{} {} {} at {}: {}",
                     job.id,
                     job.kind,
                     job.scope,
+                    job.finished_at.as_deref().unwrap_or(&job.started_at),
                     job.error.unwrap_or_else(|| "failed".to_string())
                 )
             })
@@ -517,6 +516,28 @@ fn recent_jobs_check(database: Option<&Database>) -> Result<DoctorCheck, Execute
     ))
 }
 
+fn repair_for_job_subject(subject: &JobDiagnosticSubject) -> String {
+    match subject {
+        JobDiagnosticSubject::UpdateAssessment => "pv update".to_owned(),
+        JobDiagnosticSubject::Resource { name, track: _ } if name == "composer" => {
+            "pv composer:install".to_owned()
+        }
+        JobDiagnosticSubject::Resource { name, track }
+            if matches!(
+                name.as_str(),
+                "mailpit" | "mysql" | "postgres" | "redis" | "rustfs"
+            ) =>
+        {
+            format!("pv {name}:install {track}")
+        }
+        JobDiagnosticSubject::SystemReconciliation
+        | JobDiagnosticSubject::GatewayRuntime
+        | JobDiagnosticSubject::Project { .. }
+        | JobDiagnosticSubject::Resource { .. }
+        | JobDiagnosticSubject::Other { .. } => "pv daemon:restart".to_owned(),
+    }
+}
+
 fn runtime_states_check(database: Option<&Database>) -> Result<DoctorCheck, ExecuteError> {
     let Some(database) = database else {
         return Ok(DoctorCheck::warn(

crates/cli/src/commands/ports.rs

@@ -4,7 +4,10 @@ use std::process::ExitCode;
 
 use camino::{Utf8Path, Utf8PathBuf};
 use platform::{PfConfReference, PfFileState, PfRedirectConfig};
-use state::{Database, GatewayPort, GatewayPortAssignments, PortOwner, PvPaths, StateError};
+use state::{
+    Database, GatewayPort, GatewayPortAssignments, PortOwner, PvPaths, RuntimeObservedStatus,
+    RuntimeSubject, StateError,
+};
 
 use crate::args::PortsStatusArgs;
 use crate::environment::Environment;
@@ -159,6 +162,12 @@ pub(crate) fn install(
 
         if active_config.as_ref() == Some(&config) {
             output.line("System pf redirect config already matches PV")?;
+            refresh_gateway_observation_after_pf_repair(
+                environment,
+                &paths,
+                &config,
+                &mut database,
+            )?;
 
             return Ok(ExitCode::SUCCESS);
         }
@@ -183,11 +192,54 @@ pub(crate) fn install(
         had_http_assignment,
         had_https_assignment,
     )?;
+    refresh_gateway_observation_after_pf_repair(environment, &paths, &config, &mut database)?;
     output.line("Installed system pf redirect config")?;
 
     Ok(ExitCode::SUCCESS)
 }
 
+fn refresh_gateway_observation_after_pf_repair(
+    environment: &impl Environment,
+    paths: &PvPaths,
+    config: &PfRedirectConfig,
+    database: &mut Database,
+) -> Result<(), ExecuteError> {
+    if environment
+        .probe_gateway_redirects(config, &paths.ca_certificate())
+        .is_ok()
+    {
+        database.record_runtime_observed_snapshot(
+            RuntimeSubject::Gateway,
+            RuntimeObservedStatus::Running,
+            Some("Gateway identity verified through ports 80 and 443 after PF repair"),
+        )?;
+
+        return Ok(());
+    }
+
+    let pf_derived_observation = database
+        .runtime_observed_states()?
+        .into_iter()
+        .any(|state| {
+            state.subject == RuntimeSubject::Gateway
+                && state.status == RuntimeObservedStatus::Degraded
+                && state
+                    .message
+                    .as_deref()
+                    .is_some_and(|message| message.starts_with("Low-port routing is "))
+        });
+    if pf_derived_observation {
+        database.record_runtime_observed_snapshot(
+            RuntimeSubject::Gateway,
+            RuntimeObservedStatus::Pending,
+            Some("Low-port routing repaired; Gateway readiness is pending reconciliation"),
+        )?;
+    }
+    let _request_result = environment.request_system_reconciliation(paths);
+
+    Ok(())
+}
+
 fn ensure_active_gateway_ports(
     environment: &impl Environment,
     config: &PfRedirectConfig,

crates/cli/src/commands/status.rs

@@ -8,7 +8,7 @@ use platform::{
 };
 use serde::Serialize;
 use state::{
-    Database, JobRecord, JobStatus, ManagedResourceDesiredState, ManagedResourceTrackRecord,
+    Database, JobRecord, ManagedResourceDesiredState, ManagedResourceTrackRecord,
     ProjectEnvObservedStatus, ProjectRecord, PvPaths, RuntimeObservedStateRecord,
     RuntimeObservedStatus, RuntimeSubject, StateError,
 };
@@ -83,10 +83,9 @@ impl StatusSnapshot {
         };
         let recent_errors = match &database {
             Some(database) => database
-                .recent_jobs()?
+                .unresolved_job_failures()?
                 .into_iter()
-                .filter(|job| job.status == JobStatus::Failed)
-                .map(JobStatusSummary::from_job)
+                .map(|failure| JobStatusSummary::from_job(failure.job))
                 .collect::<Vec<_>>(),
             None => Vec::new(),
         };
@@ -176,10 +175,11 @@ impl StatusSnapshot {
         } else {
             for job in &self.recent_errors {
                 output.line(&format!(
-                    "  {} {} {} failed: {}",
+                    "  {} {} {} failed at {}: {}",
                     job.id,
                     job.kind,
                     job.scope,
+                    job.finished_at.as_deref().unwrap_or(&job.started_at),
                     job.error.as_deref().unwrap_or("-"),
                 ))?;
             }

crates/cli/src/environment.rs

@@ -141,6 +141,10 @@ pub trait Environment {
         Err("Gateway identity probing is unavailable in this environment".to_owned())
     }
 
+    fn request_system_reconciliation(&self, _paths: &state::PvPaths) -> Result<(), String> {
+        Err("daemon reconciliation requests are unavailable in this environment".to_owned())
+    }
+
     fn remove_pf_redirects(
         &self,
         system_anchor_path: &Utf8Path,
@@ -258,6 +262,12 @@ impl Environment for ProcessEnvironment {
             .map_err(|error| error.to_string())
     }
 
+    fn request_system_reconciliation(&self, paths: &state::PvPaths) -> Result<(), String> {
+        daemon::submit_job_blocking(paths.clone(), "reconcile", "system")
+            .map(|_job| ())
+            .map_err(|error| error.to_string())
+    }
+
     fn exec(&self, program: &Path, args: &[String]) -> io::Result<ExitCode> {
         self.exec_with_env(program, args, &[])
     }

crates/cli/tests/doctor.rs

@@ -15,7 +15,7 @@ use platform::{
     ActivePfRedirectInspection, KeychainCertificate, KeychainTrustResult, LaunchAgentConfig,
     PfConfReference, PfRedirectConfig, ResolverConfig,
 };
-use state::{Database, PvPaths, RuntimeObservedStatus, RuntimeSubject};
+use state::{Database, JobDiagnosticSubject, PvPaths, RuntimeObservedStatus, RuntimeSubject};
 
 #[derive(Debug)]
 struct TestEnvironment {
@@ -191,6 +191,49 @@ fn doctor_fails_with_repair_commands() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[test]
+fn doctor_tracks_failure_repair_and_identical_recurrence() -> anyhow::Result<()> {
+    let tempdir = tempdir()?;
+    let home = tempdir.path().join("home");
+    let paths = PvPaths::for_home(home.clone());
+    let environment = TestEnvironment::new(&home);
+    seed_required_checks(&paths, &environment, true)?;
+    let mut database = Database::open(&paths)?;
+    let failure = database.start_job("reconcile", "project:project_1")?;
+    database.fail_job(&failure.id, "Gateway failed to start")?;
+
+    let failed = run_doctor_with_health(&paths, &environment)?;
+    let repair = database.start_job("reconcile", "project:project_1")?;
+    database.complete_job_with_coverage(
+        &repair.id,
+        "Project reconciled",
+        &[
+            JobDiagnosticSubject::Project {
+                id: "project_1".to_owned(),
+            },
+            JobDiagnosticSubject::GatewayRuntime,
+        ],
+    )?;
+    let healthy = run_doctor_with_health(&paths, &environment)?;
+    let recurrence = database.start_job("reconcile", "project:project_1")?;
+    database.fail_job(&recurrence.id, "Gateway failed to start")?;
+    let recurring = run_doctor_with_health(&paths, &environment)?;
+
+    assert_eq!(failed.exit_code, ExitCode::FAILURE);
+    assert_eq!(healthy.exit_code, ExitCode::SUCCESS);
+    assert_eq!(recurring.exit_code, ExitCode::FAILURE);
+    assert!(failed.stderr.is_empty());
+    assert!(healthy.stderr.is_empty());
+    assert!(recurring.stderr.is_empty());
+    assert_doctor_snapshot(
+        "doctor_tracks_failure_repair_and_identical_recurrence",
+        tempdir.path(),
+        (failed, healthy, recurring),
+    );
+
+    Ok(())
+}
+
 #[test]
 fn doctor_fails_when_daemon_socket_is_stale() -> anyhow::Result<()> {
     let tempdir = tempdir()?;
@@ -345,6 +388,20 @@ fn run_pv(args: &[&str], environment: &impl Environment) -> anyhow::Result<RunOu
     })
 }
 
+fn run_doctor_with_health(
+    paths: &PvPaths,
+    environment: &impl Environment,
+) -> anyhow::Result<RunOutput> {
+    if state::fs::path_exists(&paths.daemon_socket()) {
+        state::fs::delete_file(&paths.daemon_socket())?;
+    }
+    let health_server = spawn_health_server(&paths.daemon_socket())?;
+    let output = run_pv(&["doctor"], environment)?;
+    join_health_server(health_server)?;
+
+    Ok(output)
+}
+
 fn seed_required_checks(
     paths: &PvPaths,
     environment: &TestEnvironment,