Centralize operator liveness state

apps/decodex-app/Sources/DecodexApp/AccountPanelView.swift

@@ -2954,7 +2954,7 @@ struct OperatorLanePopoverView: View {
 	}
 
 	private var currentSummary: String {
-		if run.processAlive == false {
+		if run.processAlive == false, run.hasFreshExecution == false {
 			if let idle = formatActivityDuration(run.inactiveDurationSeconds) {
 				return "Stopped · idle \(idle)"
 			}

apps/decodex-app/Sources/DecodexApp/OperatorSnapshotModels.swift

@@ -1,5 +1,7 @@
 import Foundation
 
+private let operatorActiveRunIdleTimeoutSeconds = 300
+
 struct OperatorSnapshotResponse: Decodable, Sendable {
 	let warnings: [String]
 	let projects: [OperatorProjectStatus]
@@ -30,13 +32,13 @@ struct OperatorSnapshotResponse: Decodable, Sendable {
 
 	var reviewCount: Int {
 		max(
-			postReviewLanes.count,
+			postReviewLanes.filter { $0.shadowedByActiveRun == false }.count,
 			projects.reduce(0) { $0 + $1.postReviewLaneCount }
 		)
 	}
 
 	var landingCount: Int {
-		postReviewLanes.filter { $0.isReadyToLand }.count
+		postReviewLanes.filter { $0.isReadyToLand && $0.shadowedByActiveRun == false }.count
 	}
 
 	var waitingCount: Int {
@@ -298,13 +300,21 @@ struct OperatorQueuedIssueStatus: Decodable, Sendable {
 
 struct OperatorPostReviewLaneStatus: Decodable, Sendable {
 	let classification: String?
+	let shadowedByActiveRun: Bool
 
 	var isReadyToLand: Bool {
-		classification == "ready_to_land"
+		classification == "ready_to_land" && shadowedByActiveRun == false
 	}
 
 	enum CodingKeys: String, CodingKey {
 		case classification
+		case shadowedByActiveRun = "shadowed_by_active_run"
+	}
+
+	init(from decoder: Decoder) throws {
+		let container = try decoder.container(keyedBy: CodingKeys.self)
+		classification = try container.decodeIfPresent(String.self, forKey: .classification)
+		shadowedByActiveRun = try container.decodeIfPresent(Bool.self, forKey: .shadowedByActiveRun) ?? false
 	}
 }
 
@@ -420,14 +430,20 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 	let waitReason: String?
 	let currentOperation: String?
 	let threadStatus: String?
+	let threadActiveFlags: [String]
 	let idleForSeconds: Int?
 	let protocolIdleForSeconds: Int?
 	let updatedAt: String?
 	let lastProgressAt: String?
 	let nextRetryAt: String?
 	let lastEventType: String?
 	let eventCount: Int?
+	let executionLiveness: String?
+	let hasFreshExecutionSnapshot: Bool?
+	let countsAsRunningSnapshot: Bool?
+	let needsAttentionSnapshot: Bool?
 	let processAlive: Bool?
+	let processLivenessReason: String?
 	let activeLease: Bool?
 	let branchName: String?
 	let worktreePath: String?
@@ -511,11 +527,15 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 	}
 
 	var hasAttentionTone: Bool {
-		suspectedStall
+		if let needsAttentionSnapshot {
+			return needsAttentionSnapshot
+		}
+
+		return suspectedStall
 			|| attemptStatus == "waiting_for_review"
 			|| status == "manual_attention"
 			|| status == "blocked"
-			|| processAlive == false
+			|| stoppedProcessNeedsAttention
 	}
 
 	var isWaiting: Bool {
@@ -542,18 +562,39 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 	var shouldRetainDuringPartialRunActivity: Bool {
 		activeLease == true
 			|| processAlive == true
+			|| hasRecentAppServerExecution
 			|| status == "running"
 			|| phase == "executing"
 	}
 
 	var countsAsRunning: Bool {
-		hasRunningStatus
+		if let countsAsRunningSnapshot {
+			return countsAsRunningSnapshot
+		}
+
+		return hasRunningStatus
 			&& phase == "executing"
-			&& processAlive != false
+			&& (processAlive != false || hasFreshExecution)
 			&& hasAttentionTone == false
 			&& hasStaleExecutionWithoutKnownProcess == false
 	}
 
+	var hasFreshExecution: Bool {
+		if let hasFreshExecutionSnapshot {
+			return hasFreshExecutionSnapshot
+		}
+
+		return hasRunningStatus
+			&& (processAlive == true || hasRecentAppServerExecution)
+	}
+
+	private var hasRecentAppServerExecution: Bool {
+		threadStatus == "active"
+			|| threadActiveFlags.isEmpty == false
+			|| ["thread_active", "protocol_observed"].contains(executionLiveness ?? "")
+			|| protocolIdleForSeconds.isSomeAndLessThan(operatorActiveRunIdleTimeoutSeconds)
+	}
+
 	private var hasRunningStatus: Bool {
 		guard let status else {
 			return false
@@ -562,6 +603,13 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 		return ["starting", "running"].contains(status)
 	}
 
+	private var stoppedProcessNeedsAttention: Bool {
+		processAlive == false
+			&& hasRunningStatus
+			&& waitReason == nil
+			&& hasFreshExecution == false
+	}
+
 	private var hasStaleExecutionWithoutKnownProcess: Bool {
 		hasRunningStatus
 			&& phase == "executing"
@@ -591,14 +639,20 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 			waitReason: activity.waitReason ?? waitReason,
 			currentOperation: activity.currentOperation ?? currentOperation,
 			threadStatus: activity.threadStatus ?? threadStatus,
+			threadActiveFlags: activity.threadActiveFlags.isEmpty ? threadActiveFlags : activity.threadActiveFlags,
 			idleForSeconds: activity.idleForSeconds ?? idleForSeconds,
 			protocolIdleForSeconds: activity.protocolIdleForSeconds ?? protocolIdleForSeconds,
 			updatedAt: activity.updatedAt ?? updatedAt,
 			lastProgressAt: activity.lastProgressAt ?? lastProgressAt,
 			nextRetryAt: activity.nextRetryAt ?? nextRetryAt,
 			lastEventType: activity.lastEventType ?? lastEventType,
 			eventCount: activity.eventCount ?? eventCount,
+			executionLiveness: activity.executionLiveness ?? executionLiveness,
+			hasFreshExecutionSnapshot: activity.hasFreshExecutionSnapshot ?? hasFreshExecutionSnapshot,
+			countsAsRunningSnapshot: activity.countsAsRunningSnapshot ?? countsAsRunningSnapshot,
+			needsAttentionSnapshot: activity.needsAttentionSnapshot ?? needsAttentionSnapshot,
 			processAlive: activity.processAlive ?? processAlive,
+			processLivenessReason: activity.processLivenessReason ?? processLivenessReason,
 			activeLease: activity.activeLease ?? activeLease,
 			branchName: activity.branchName ?? branchName,
 			worktreePath: activity.worktreePath ?? worktreePath,
@@ -640,14 +694,20 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 		case waitReason = "wait_reason"
 		case currentOperation = "current_operation"
 		case threadStatus = "thread_status"
+		case threadActiveFlags = "thread_active_flags"
 		case idleForSeconds = "idle_for_seconds"
 		case protocolIdleForSeconds = "protocol_idle_for_seconds"
 		case updatedAt = "updated_at"
 		case lastProgressAt = "last_progress_at"
 		case nextRetryAt = "next_retry_at"
 		case lastEventType = "last_event_type"
 		case eventCount = "event_count"
+		case executionLiveness = "execution_liveness"
+		case hasFreshExecutionSnapshot = "has_fresh_execution"
+		case countsAsRunningSnapshot = "counts_as_running"
+		case needsAttentionSnapshot = "needs_attention"
 		case processAlive = "process_alive"
+		case processLivenessReason = "process_liveness_reason"
 		case activeLease = "active_lease"
 		case branchName = "branch_name"
 		case worktreePath = "worktree_path"
@@ -676,14 +736,20 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 		waitReason = try container.decodeIfPresent(String.self, forKey: .waitReason)
 		currentOperation = try container.decodeIfPresent(String.self, forKey: .currentOperation)
 		threadStatus = try container.decodeIfPresent(String.self, forKey: .threadStatus)
+		threadActiveFlags = try container.decodeIfPresent([String].self, forKey: .threadActiveFlags) ?? []
 		idleForSeconds = try container.decodeIfPresent(Int.self, forKey: .idleForSeconds)
 		protocolIdleForSeconds = try container.decodeIfPresent(Int.self, forKey: .protocolIdleForSeconds)
 		updatedAt = try container.decodeIfPresent(String.self, forKey: .updatedAt)
 		lastProgressAt = try container.decodeIfPresent(String.self, forKey: .lastProgressAt)
 		nextRetryAt = try container.decodeIfPresent(String.self, forKey: .nextRetryAt)
 		lastEventType = try container.decodeIfPresent(String.self, forKey: .lastEventType)
 		eventCount = try container.decodeIfPresent(Int.self, forKey: .eventCount)
+		executionLiveness = try container.decodeIfPresent(String.self, forKey: .executionLiveness)
+		hasFreshExecutionSnapshot = try container.decodeIfPresent(Bool.self, forKey: .hasFreshExecutionSnapshot)
+		countsAsRunningSnapshot = try container.decodeIfPresent(Bool.self, forKey: .countsAsRunningSnapshot)
+		needsAttentionSnapshot = try container.decodeIfPresent(Bool.self, forKey: .needsAttentionSnapshot)
 		processAlive = try container.decodeIfPresent(Bool.self, forKey: .processAlive)
+		processLivenessReason = try container.decodeIfPresent(String.self, forKey: .processLivenessReason)
 		activeLease = try container.decodeIfPresent(Bool.self, forKey: .activeLease)
 		branchName = try container.decodeIfPresent(String.self, forKey: .branchName)
 		worktreePath = try container.decodeIfPresent(String.self, forKey: .worktreePath)
@@ -714,14 +780,20 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 		waitReason: String?,
 		currentOperation: String?,
 		threadStatus: String?,
+		threadActiveFlags: [String],
 		idleForSeconds: Int?,
 		protocolIdleForSeconds: Int?,
 		updatedAt: String?,
 		lastProgressAt: String?,
 		nextRetryAt: String?,
 		lastEventType: String?,
 		eventCount: Int?,
+		executionLiveness: String?,
+		hasFreshExecutionSnapshot: Bool?,
+		countsAsRunningSnapshot: Bool?,
+		needsAttentionSnapshot: Bool?,
 		processAlive: Bool?,
+		processLivenessReason: String?,
 		activeLease: Bool?,
 		branchName: String?,
 		worktreePath: String?,
@@ -744,14 +816,20 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 		self.waitReason = waitReason
 		self.currentOperation = currentOperation
 		self.threadStatus = threadStatus
+		self.threadActiveFlags = threadActiveFlags
 		self.idleForSeconds = idleForSeconds
 		self.protocolIdleForSeconds = protocolIdleForSeconds
 		self.updatedAt = updatedAt
 		self.lastProgressAt = lastProgressAt
 		self.nextRetryAt = nextRetryAt
 		self.lastEventType = lastEventType
 		self.eventCount = eventCount
+		self.executionLiveness = executionLiveness
+		self.hasFreshExecutionSnapshot = hasFreshExecutionSnapshot
+		self.countsAsRunningSnapshot = countsAsRunningSnapshot
+		self.needsAttentionSnapshot = needsAttentionSnapshot
 		self.processAlive = processAlive
+		self.processLivenessReason = processLivenessReason
 		self.activeLease = activeLease
 		self.branchName = branchName
 		self.worktreePath = worktreePath
@@ -763,6 +841,16 @@ struct OperatorRunStatus: Decodable, Identifiable, Sendable {
 	}
 }
 
+private extension Optional where Wrapped == Int {
+	func isSomeAndLessThan(_ threshold: Int) -> Bool {
+		guard let value = self else {
+			return false
+		}
+
+		return value < threshold
+	}
+}
+
 struct OperatorDashboardSocketEvent: Decodable, Sendable {
 	let type: String
 	let payload: OperatorDashboardSocketPayload?

apps/decodex-app/Tests/DecodexAppTests/AccountModelTests.swift

@@ -283,6 +283,36 @@ final class AccountModelTests: XCTestCase {
 		XCTAssertEqual(run.inactiveDurationSeconds, 20840)
 	}
 
+	func testFreshProtocolExecutionOverridesStaleProcessMarker() throws {
+		let payload = """
+		{
+		  "run_id": "xy-957-attempt-3",
+		  "issue_identifier": "XY-957",
+		  "status": "running",
+		  "phase": "executing",
+		  "wait_reason": "model_execution",
+		  "execution_liveness": "process_identity_mismatch",
+		  "has_fresh_execution": true,
+		  "counts_as_running": true,
+		  "needs_attention": false,
+		  "process_alive": false,
+		  "process_liveness_reason": "host_boot_id_mismatch",
+		  "thread_status": "active",
+		  "idle_for_seconds": 1,
+		  "protocol_idle_for_seconds": 1,
+		  "last_progress_at": "2026-06-16T03:27:31Z",
+		  "last_event_type": "turn/diff/updated"
+		}
+		""".data(using: .utf8)!
+
+		let run = try JSONDecoder().decode(OperatorRunStatus.self, from: payload)
+
+		XCTAssertTrue(run.hasFreshExecution)
+		XCTAssertTrue(run.countsAsRunning)
+		XCTAssertFalse(run.hasAttentionTone)
+		XCTAssertEqual(run.inactiveDurationSeconds, 1)
+	}
+
 	func testOperatorSnapshotAssignsCodexAccountRunsToAccountRows() throws {
 		let assignedAccount = makeAccount(
 			status: "available",
@@ -423,6 +453,31 @@ final class AccountModelTests: XCTestCase {
 		XCTAssertEqual(snapshot.runningLaneCount, 2)
 	}
 
+	func testShadowedPostReviewLaneDoesNotInflateReviewOrLandingCounts() throws {
+		let payload = """
+		{
+		  "projects": [
+		    {
+		      "project_id": "pubfi-platform",
+		      "post_review_lane_count": 0
+		    }
+		  ],
+		  "post_review_lanes": [
+		    {
+		      "classification": "ready_to_land",
+		      "shadowed_by_active_run": true
+		    }
+		  ]
+		}
+		""".data(using: .utf8)!
+
+		let snapshot = try JSONDecoder().decode(OperatorSnapshotResponse.self, from: payload)
+
+		XCTAssertEqual(snapshot.reviewCount, 0)
+		XCTAssertEqual(snapshot.landingCount, 0)
+		XCTAssertEqual(snapshot.postReviewLanes.first?.shadowedByActiveRun, true)
+	}
+
 	func testOperatorSnapshotAssignsSelectedAccountWhenPrimaryAccountIsMissing() throws {
 		let assignedAccount = makeAccount(
 			status: "available",