SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Стабилизация восстановления Bridge и Quick Connect

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-23 10:45:53 +03:00
parent 71c8b043c9
commit 7d911b4b1f
6 changed files with 381 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/control-station/test/k1SupervisorPresentation.test.mjs
+14
View File
@@ -6185,6 +6185,20 @@ test("Apply presentation stays correlated through ACK, bootstrap, terminal and c
"settling",
"ready child success waits for the exact reachable lease projection",
);
assert.equal(
provisioningAttemptViewState(correlated, {
...initialState,
operations: [operation],
connection_attempt: {
...attempt,
status: "succeeded",
control_state: "unknown",
safe_next_action: "verify-control-read-only",
},
}),
"settling",
"the network-success/control-bootstrap handoff keeps the existing loader",
);
assert.equal(
provisioningAttemptViewState(correlated, {
...initialState,
plugins/xgrids-k1/frontend/src/components/K1ProvisioningPipeline.tsx
+12
View File
@@ -351,6 +351,18 @@ export function provisioningAttemptViewState(
if (attempt.status === "succeeded" && attempt.control_state === "ready") {
return "settling";
}
// The network operation can be projected as succeeded immediately before
// its service-owned control bootstrap appears in the next snapshot. Keep
// the click-owned inline progress surface during that exact handoff; a
// terminal bootstrap failure still arrives as failed/control_not_ready.
if (
attempt.status === "succeeded"
&& attempt.phase === "network_applied"
&& attempt.control_state === "unknown"
&& attempt.safe_next_action === "verify-control-read-only"
) {
return "settling";
}
if (TERMINAL_PROVISIONING_STATUSES.has(attempt.status)) return "failed";
}
if (
src/k1link/device_plugins/xgrids_k1/facade.py
+150 -17
View File
@@ -12885,6 +12885,8 @@ class XgridsK1CompatibilityService:
def _validate_active_acquisition_checkpoint_lineage(
self,
*,
allow_untrusted_terminal_settlement: bool = False,
) -> _ActiveAcquisitionCheckpointTrustToken | None:
"""Recompute the current unceased checkpoint/ledger lineage locally.
@@ -12892,13 +12894,23 @@ class XgridsK1CompatibilityService:
ledger and immutable archive retain its proof material. This method is
deliberately read-only: a mismatch revokes local START/resume admission
for the process lifetime and never repairs either durable document.
``allow_untrusted_terminal_settlement`` is reserved for one narrower
caller which has already durably committed an exact, non-retained
READY/SCAN_OVER reconciliation. It may recompute a token only to cease
the old checkpoint; process-local START/resume trust stays revoked until
that terminal CAS has succeeded.
"""
if getattr(
current_trust = getattr(
self,
"_active_acquisition_checkpoint_trust",
"trusted",
) != "trusted":
)
if current_trust != "trusted" and not (
allow_untrusted_terminal_settlement
and current_trust == "unavailable"
):
return None
store = self._active_acquisition_checkpoint
if store is None:
@@ -13023,6 +13035,110 @@ class XgridsK1CompatibilityService:
)
return None
def _settle_untrusted_checkpoint_after_verified_standby(
self,
*,
reconciliation_id: str,
reconciled_record: Mapping[str, Any],
) -> bool:
"""Consume a new durable standby proof without reviving START authority.
A process can reject an ACTIVE checkpoint while an older STOP is still
unresolved. If the same process later appends an exact read-only
READY/SCAN_OVER reconciliation, that durable change is new evidence.
Recompute the lineage only for terminal checkpoint cessation; no K1
command, network mutation, runtime rehydration, or acquisition authority
is permitted by this path.
"""
if getattr(
self,
"_active_acquisition_checkpoint_trust",
"trusted",
) != "unavailable":
return False
store = self._active_acquisition_checkpoint
if store is None:
return False
try:
checkpoint_snapshot = store.snapshot()
ledger_snapshot = self._physical_command_ledger.snapshot()
record = ledger_snapshot.record
reconciliation = (
record.reconciliations[-1]
if record is not None and record.reconciliations
else None
)
if not (
checkpoint_snapshot.status in {"prepared", "active"}
and checkpoint_snapshot.checkpoint is not None
and ledger_snapshot.status == "resolved"
and record is not None
and record.as_dict() == dict(reconciled_record)
and reconciliation is not None
and reconciliation.reconciliation_id == reconciliation_id
and reconciliation.resolution == "physical-standby-observed"
and reconciliation.kind
in {
"ambiguous-outcome",
"prepared-stop-classification",
"resolved-active-cessation",
}
and reconciliation.observation.source
== "explicit-read-only-reconciliation"
and reconciliation.observation.session_state
in {"ready", "scan_over"}
and not reconciliation.observation.project_bound
and not reconciliation.observation.init_ready
and not reconciliation.observation.mqtt_retained
):
return False
token = self._validate_active_acquisition_checkpoint_lineage(
allow_untrusted_terminal_settlement=True,
)
if token is None:
return False
settled = self._settle_restart_checkpoint_after_verified_standby(
token=token,
reconciliation_id=reconciliation_id,
reconciled_record=reconciled_record,
)
if not settled:
return False
committed = store.snapshot().checkpoint
if not (
committed is not None
and committed.state == "ceased"
and committed.acquisition_id == token.acquisition_id
and committed.original_start_operation_id
== token.root_start_operation_id
):
return False
with self._lock:
if self._active_acquisition_checkpoint_trust == "unavailable":
self._active_acquisition_checkpoint_trust = "trusted"
self._active_acquisition_checkpoint_reason = None
logger.info(
"untrusted K1 checkpoint closed from new durable standby proof",
extra={
"event_code": (
"active_acquisition_checkpoint_untrusted_standby_settled"
),
"acquisition_id": token.acquisition_id,
"reconciliation_id": reconciliation_id,
"device_write_performed": False,
"automatic_retry": False,
},
)
return True
except (
ActiveAcquisitionRecoveryCheckpointError,
PhysicalCommandLedgerError,
OSError,
ValueError,
):
return False
def _require_active_acquisition_checkpoint_store(
self,
) -> ActiveAcquisitionRecoveryCheckpointStore:
@@ -16105,8 +16221,7 @@ class XgridsK1CompatibilityService:
else None
)
checkpoint_settlement_required = bool(
checkpoint_trust_token is not None
and ledger_snapshot.status == "resolved"
ledger_snapshot.status == "resolved"
and record is not None
and latest_reconciliation is not None
and latest_reconciliation.resolution
@@ -16128,14 +16243,24 @@ class XgridsK1CompatibilityService:
if checkpoint_settlement_required:
assert record is not None
assert latest_reconciliation is not None
settled = self._settle_restart_checkpoint_after_verified_standby(
token=checkpoint_trust_token,
reconciliation_id=(
latest_reconciliation.reconciliation_id
),
reconciled_record=record.as_dict(),
)
if not settled:
if checkpoint_trust_token is not None:
settled = self._settle_restart_checkpoint_after_verified_standby(
token=checkpoint_trust_token,
reconciliation_id=(
latest_reconciliation.reconciliation_id
),
reconciled_record=record.as_dict(),
)
else:
settled = (
self._settle_untrusted_checkpoint_after_verified_standby(
reconciliation_id=(
latest_reconciliation.reconciliation_id
),
reconciled_record=record.as_dict(),
)
)
if checkpoint_trust_token is not None and not settled:
raise ConnectionVerificationError(
"Подтверждённый READY не закрыл старую локальную START-сессию",
reason_code=(
@@ -16369,11 +16494,19 @@ class XgridsK1CompatibilityService:
checkpoint_lineage=checkpoint_lineage,
)
elif reconciled_record is not None:
settled = self._settle_restart_checkpoint_after_verified_standby(
token=checkpoint_trust_token,
reconciliation_id=reconciliation_id,
reconciled_record=reconciled_record,
)
if checkpoint_trust_token is not None:
settled = self._settle_restart_checkpoint_after_verified_standby(
token=checkpoint_trust_token,
reconciliation_id=reconciliation_id,
reconciled_record=reconciled_record,
)
else:
settled = (
self._settle_untrusted_checkpoint_after_verified_standby(
reconciliation_id=reconciliation_id,
reconciled_record=reconciled_record,
)
)
if checkpoint_trust_token is not None and not settled:
raise ActiveAcquisitionRecoveryCheckpointError(
"restart standby checkpoint settlement failed closed"
src/k1link/web/app.py
+1
View File
@@ -535,6 +535,7 @@ _CLOSED_WEBSOCKET_SEND_ERRORS = frozenset(
"Unexpected ASGI message 'websocket.send', after sending "
"'websocket.close' or response already completed."
),
"Unexpected ASGI message 'websocket.send', after sending 'websocket.close'.",
}
)
tests/test_web_validation_security.py
+4
View File
@@ -202,6 +202,10 @@ def test_plugin_expected_state_preserves_its_non_gateway_http_status(
"unable to perform operation on <TCPTransport closed=True>; "
"the handler is closed"
),
RuntimeError(
"Unexpected ASGI message 'websocket.send', after sending "
"'websocket.close'."
),
],
)
def test_device_plugin_events_treats_proven_transport_disconnect_as_completion(
tests/test_xgrids_acquisition_lifecycle.py
+200
View File
@@ -31256,6 +31256,87 @@ def _persist_same_runtime_prepared_checkpoint_for_resolved_start(
)
def _persist_active_checkpoint_for_resolved_start(
service: XgridsK1CompatibilityService,
*,
connection_mode: facade_module.ConnectionMode,
target_ipv4: str,
) -> None:
"""Persist one proven active acquisition without process-local ownership."""
_persist_resolved_active_start_for_restart(
service,
connection_mode=connection_mode,
target_ipv4=target_ipv4,
)
record = service._physical_command_ledger.snapshot().record # noqa: SLF001
store = service._active_acquisition_checkpoint # noqa: SLF001
assert record is not None and record.last_status is not None
assert store is not None
connection = record.connection
binding = ActiveAcquisitionRecoveryTransportBinding(
runtime_instance_id=service._snapshot_runtime_id, # noqa: SLF001
intent_id=connection.intent_id,
transport_ref=connection.transport_ref,
connection_mode=connection.connection_mode,
target_ipv4=connection.target_ipv4,
target_port=connection.target_port,
host_path_epoch=connection.host_path_epoch,
control_session_id=connection.control_session_id,
producer_generation=connection.producer_generation,
logical_device_id="known-k1",
compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
vendor_device_id_sha256=_RECOVERY_VENDOR_HASH,
device_serial_sha256=_RECOVERY_SERIAL_HASH,
)
prepared = store.prepare(
transition_id="prepare-active-recovery-start",
predecessor_revision=0,
acquisition_id=record.acquisition_id,
original_start_operation_id=record.operation_id,
start_payload_sha256=record.payload_sha256,
identity=ActiveAcquisitionRecoveryIdentity(
logical_device_id="known-k1",
vendor_device_id_sha256=_RECOVERY_VENDOR_HASH,
device_serial_sha256=_RECOVERY_SERIAL_HASH,
),
connection=ActiveAcquisitionRecoveryConnection(
transport_ref=connection.transport_ref,
connection_mode=connection.connection_mode,
target_ipv4=connection.target_ipv4,
target_port=connection.target_port,
),
compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
project_name="ACTIVE_RECOVERY",
project_name_wire_sha256=active_acquisition_project_name_sha256(
"ACTIVE_RECOVERY"
),
original_evidence_session_id="evidence-active-recovery",
duration_seconds=None,
requested_streams=("spatial.point-cloud.live", "camera.rgb.live"),
evidence_policy="required",
mount_type="handheld",
gnss_mode="none",
prepared_binding=binding,
)
store.activate(
transition_id="activate-active-recovery-start",
expected_revision=prepared.revision,
expected_acquisition_id=prepared.acquisition_id,
expected_start_operation_id=prepared.original_start_operation_id,
status_proof=service._checkpoint_status_proof( # noqa: SLF001
status=record.last_status,
binding=binding,
evidence_session_id="evidence-active-recovery",
),
physical_proof=service._checkpoint_physical_proof( # noqa: SLF001
record=record,
binding=binding,
checkpoint=prepared,
),
)
def _persist_resolved_unclassified_stop_for_restart(
service: XgridsK1CompatibilityService,
*,
@@ -32930,6 +33011,125 @@ def test_restart_auto_settles_durable_prepared_resolved_start_standby(
assert restarted_runtime.stop_calls == 0
def test_new_bridge_ready_settles_untrusted_quick_checkpoint_without_command(
tmp_path: Path,
) -> None:
"""A new exact READY closes stale Quick state locally in the same process."""
service, runtime = service_with_fake_runtime(tmp_path)
_persist_active_checkpoint_for_resolved_start(
service,
connection_mode="quick-connect",
target_ipv4="192.168.56.1",
)
ledger = service._physical_command_ledger # noqa: SLF001
start = ledger.snapshot().record
assert start is not None and start.last_status is not None
stop_operation_id = "physical-stop-untrusted-quick-to-bridge"
ledger.prepare(
operation_id=stop_operation_id,
parent_operation_id=start.operation_id,
acquisition_id=start.acquisition_id,
action="stop",
identity=start.identity,
connection=start.connection,
compatibility_profile_id=start.compatibility_profile_id,
payload_sha256="6" * 64,
baseline_status=start.last_status,
)
ledger.mark_dispatching(stop_operation_id)
ledger.mark_observing(
stop_operation_id,
publish_call_returned=True,
packet_id=42,
)
ledger.mark_qos2_completed(stop_operation_id, packet_id=42)
ledger.record_application_response(
stop_operation_id,
PhysicalCommandApplicationResponse(
operation_id=stop_operation_id,
action="stop",
control_session_id=start.connection.control_session_id,
host_path_epoch=start.connection.host_path_epoch,
producer_generation=start.connection.producer_generation,
result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE,
success=True,
payload_sha256="7" * 64,
observed_at_utc="2026-08-09T12:00:03.000Z",
),
)
service._mark_active_acquisition_checkpoint_untrusted( # noqa: SLF001
trust="unavailable",
reason_code="active-acquisition-recovery-checkpoint-error",
)
assert service._validate_active_acquisition_checkpoint_lineage() is None # noqa: SLF001
coordinator = service._physical_command_coordinator # noqa: SLF001
coordinator.application_response(
ApplicationMqttResponseEvidence(
operation_key="bootstrap:bridge-ready:DeviceInfoRequest",
response_topic="lixel/application/response/device_info",
payload_sha256="8" * 64,
modeling_action=None,
result_code=None,
success=None,
observed_at_utc="2026-08-09T12:01:00.000Z",
)
)
coordinator.bind_control_session(
PhysicalCommandRuntimeBinding(
vendor_device_id_sha256=_RECOVERY_VENDOR_HASH,
device_serial_sha256=_RECOVERY_SERIAL_HASH,
compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID,
intent_id="fresh-bridge-intent",
transport_ref="test-ble-transport",
connection_mode="bridge",
target_ipv4="192.168.68.52",
target_port=facade_module.CONTROL_MQTT_PORT,
host_path_epoch=2,
control_session_id="fresh-bridge-control",
producer_generation=2,
)
)
coordinator.device_status(
ApplicationMqttDeviceStatusEvidence(
vendor_device_id_sha256=_RECOVERY_VENDOR_HASH,
device_serial_sha256=_RECOVERY_SERIAL_HASH,
session_state="ready",
session_state_code=MODELING_STATE_BASE + 300,
project_bound=False,
project_id_sha256=None,
init_ready=False,
status_message_sha256="9" * 64,
mqtt_retained=False,
observed_at_utc="2026-08-09T12:01:01.000Z",
)
)
reconciliation_id = "reconcile-untrusted-quick-to-bridge-ready"
standby_record = coordinator.reconcile_unresolved(
reconciliation_id=reconciliation_id,
)
settled = service._settle_untrusted_checkpoint_after_verified_standby( # noqa: SLF001
reconciliation_id=reconciliation_id,
reconciled_record=standby_record,
)
checkpoint_store = service._active_acquisition_checkpoint # noqa: SLF001
assert checkpoint_store is not None
checkpoint = checkpoint_store.snapshot().checkpoint
assert settled is True
assert checkpoint is not None and checkpoint.state == "ceased"
assert service._active_acquisition_checkpoint_trust == "trusted" # noqa: SLF001
assert service._active_acquisition_checkpoint_reason is None # noqa: SLF001
assert standby_record["connection"]["connection_mode"] == "quick-connect"
assert standby_record["reconciliations"][-1]["verified_binding"]["connection"][
"connection_mode"
] == "bridge"
assert runtime.start_calls == []
assert runtime.stop_calls == 0
def test_transition_invalid_fences_repeated_verify_until_service_restart(
tmp_path: Path,
) -> None: