Repair live camera, zero-drive acknowledgements and cognition integration

Cargo.toml

@@ -70,6 +70,7 @@ v4l2-camera = ["http", "dep:v4l"]
 
 [dependencies]
 anyhow = "1.0"
+image = { version = "0.25", default-features = false, features = ["jpeg"] }
 clap = { version = "4.5", features = ["derive", "env"] }
 parking_lot = "0.12"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
@@ -94,6 +95,7 @@ tower-http = { version = "0.6", optional = true, features = ["cors"] }
 webrtc = { version = "0.17.1", optional = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
+libc = "0.2"
 v4l = { version = "0.14", optional = true }
 
 [dev-dependencies]

crates/leash-waveshare/src/lib.rs

@@ -172,6 +172,14 @@ pub struct GimbalCommand {
     pub acceleration: u32,
 }
 
+/// Auxiliary PWM outputs; u8 keeps each published T132 field in 0..=255.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct LightsCommand {
+    pub id: CommandId,
+    pub chassis_pwm: u8,
+    pub camera_pwm: u8,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum GimbalError {
     NonFinite,
@@ -378,6 +386,10 @@ enum ControllerCommand {
         command: GimbalCommand,
         reply: mpsc::Sender<CommandAck>,
     },
+    Lights {
+        command: LightsCommand,
+        reply: mpsc::Sender<CommandAck>,
+    },
     RequestTelemetry {
         id: CommandId,
         reply: mpsc::Sender<CommandAck>,
@@ -423,11 +435,20 @@ impl ControllerHandle {
     }
 
     pub fn submit_gimbal(&self, command: GimbalCommand) -> Result<AckTicket, SubmitError> {
+        if self.shared.estopped.load(Ordering::Acquire) { return Err(SubmitError::EStopped); }
         let (reply, receiver) = mpsc::channel();
         self.enqueue(ControllerCommand::Gimbal { command, reply })?;
         Ok(AckTicket { receiver })
     }
 
+    /// Queue a light-only write on the same serial owner as every other frame.
+    /// Applied acknowledges host write/flush, not physical lamp feedback.
+    pub fn submit_lights(&self, command: LightsCommand) -> Result<AckTicket, SubmitError> {
+        let (reply, receiver) = mpsc::channel();
+        self.enqueue(ControllerCommand::Lights { command, reply })?;
+        Ok(AckTicket { receiver })
+    }
+
     pub fn request_telemetry(&self, id: CommandId) -> Result<AckTicket, SubmitError> {
         let (reply, receiver) = mpsc::channel();
         self.enqueue(ControllerCommand::RequestTelemetry { id, reply })?;
@@ -1001,8 +1022,15 @@ fn process_command(
             )
         }
         ControllerCommand::Gimbal { command, reply } => {
+            if shared.estopped.load(Ordering::Acquire) {
+                reply_rejected(ControllerCommand::Gimbal { command, reply }, AckOutcome::EStopped, clock.now(), None);
+                return;
+            }
             (encode_gimbal(command), command.id, None, false, reply)
         }
+        ControllerCommand::Lights { command, reply } => {
+            (encode_lights(command), command.id, None, false, reply)
+        }
         ControllerCommand::RequestTelemetry { id, reply } => {
             (encode_telemetry_request(), id, None, false, reply)
         }
@@ -1053,6 +1081,7 @@ fn reply_rejected(
             (command.command_id(), Some(command.evidence_id()), reply)
         }
         ControllerCommand::Gimbal { command, reply } => (command.id, None, reply),
+        ControllerCommand::Lights { command, reply } => (command.id, None, reply),
         ControllerCommand::RequestTelemetry { id, reply } => (id, None, reply),
         ControllerCommand::EnableTelemetry { id, reply } => (id, None, reply),
     };
@@ -1159,6 +1188,14 @@ fn read_telemetry(
     Ok(())
 }
 
+pub fn encode_lights(command: LightsCommand) -> Vec<u8> {
+    let mut frame = serde_json::to_vec(&json!({
+        "T": 132, "IO4": command.chassis_pwm, "IO5": command.camera_pwm,
+    })).expect("light frame contains only integer fields");
+    frame.push(b'\n');
+    frame
+}
+
 pub fn encode_drive(command: DifferentialDrive, invert: bool, swap: bool) -> Vec<u8> {
     let (mut left, mut right) = if swap {
         (command.right.get(), command.left.get())
@@ -1363,6 +1400,39 @@ mod tests {
         }
     }
 
+    #[test]
+    fn light_commands_use_only_t132_and_acknowledge_write_not_feedback() {
+        let transcript = Arc::new(Mutex::new(Transcript { maximum_write: 2, ..Transcript::default() }));
+        let owner = owner(Arc::clone(&transcript));
+        let handle = owner.handle();
+        wait_connected(&handle);
+        lock(&transcript).writes.clear();
+        let ack = handle.submit_lights(LightsCommand {
+            id: command_id(20), chassis_pwm: 0, camera_pwm: 255,
+        }).unwrap().wait_timeout(Duration::from_secs(1)).unwrap().unwrap();
+        assert_eq!(ack.command_id, command_id(20));
+        assert_eq!(ack.outcome, AckOutcome::Applied);
+        assert!(!ack.verified_zero);
+        assert!(ack.evidence_id.is_none());
+        let frames: Vec<serde_json::Value> = String::from_utf8(lock(&transcript).writes.clone()).unwrap()
+            .lines().map(|line| serde_json::from_str(line).unwrap()).collect();
+        assert_eq!(frames, vec![json!({"T":132,"IO4":0,"IO5":255})]);
+    }
+
+    #[test]
+    fn a_failed_light_write_is_not_an_applied_lamp_state() {
+        let transcript = Arc::new(Mutex::new(Transcript::default()));
+        let owner = owner(Arc::clone(&transcript));
+        let handle = owner.handle();
+        wait_connected(&handle);
+        lock(&transcript).fail_writes = true;
+        let ack = handle.submit_lights(LightsCommand {id:command_id(21),chassis_pwm:1,camera_pwm:180})
+            .unwrap().wait_timeout(Duration::from_secs(1)).unwrap().unwrap();
+        assert_eq!(ack.outcome,AckOutcome::IoFailed);
+        assert!(ack.applied_sequence.is_none());
+        assert!(!ack.verified_zero);
+    }
+
     #[test]
     fn telemetry_enable_preserves_the_two_legacy_owner_frames() {
         let encoded = String::from_utf8(encode_telemetry_enable()).unwrap();

docs/CAMERA.md

@@ -1,9 +1,18 @@
 # Camera Capture And Streaming
 
-Leash gives snapshots, MJPEG, native V4L2 capture, and WebRTC one camera
-ownership boundary. Only one capture or stream may own the device at a time.
-Concurrent callers receive `camera is busy` instead of opening a second encoder
-or racing the active stream.
+Native V4L2 snapshots and MJPEG streams share one persistent camera owner. The
+first request starts capture; subsequent callers consume the latest actual JPEG
+through a watch channel. A slow stream skips frames without blocking capture.
+FFmpeg and WebRTC retain exclusive device ownership and cannot open a second
+device owner while native capture is active.
+
+Frames must be at most 500 ms old. When V4L2 supplies a monotonic timestamp, this
+bound includes time spent in the driver queue. Otherwise it bounds time since
+dequeue, and the snapshot explicitly reports `dequeue-only-exposure-unknown` in
+`X-Leash-Camera-Timestamp-Basis`. Snapshot headers also expose the unchanged
+driver sequence, timestamp in microseconds, timestamp flags, acquisition-start
+time and dequeue time. Requests never replace those times with a cache-hit time.
+The native worker drains the device continuously and owns no client output queue.
 
 ## Build
 
@@ -74,12 +83,53 @@ curl -s -X POST http://127.0.0.1:8000/camera/stream/recover
 ```
 
 `GET /camera/stream/health` (also `/camera/health`) reports device availability,
-the active owner (`snapshot`, `mjpeg`, or `webrtc`), start time, recovery
+the active owner (`v4l2-shared`, `snapshot`, `mjpeg`, or `webrtc`), start time, recovery
 generation/count, and the last 16 sanitized failure reasons. `/camera/status`
 includes the same object as `stream_health`.
 
-`POST /camera/stream/recover` advances the recovery generation. An active MJPEG
-or WebRTC stream observes the new generation, stops its capture process, releases
-the single-owner guard, and can then be reconnected by the operator. Recovery is
+`POST /camera/stream/recover` advances the recovery generation. The native owner,
+active FFmpeg MJPEG stream, or WebRTC stream observes the new generation, closes
+capture, and releases the owner guard. Native capture is restarted by the next
+snapshot or stream subscription after the old owner exits. Recovery is
 safe to call while idle; it does not touch motors, serial ownership, or actuation
 policy.
+
+## Camera-only gimbal authority
+
+`POST /camera/aim` (alias `/gimbal/aim`) accepts a camera-only bearer when
+`LEASH_ALLOW_CAMERA_AIM=1` is enabled on the existing Waveshare physical profile
+with its existing physical-actuation permission and policy gates. The bearer is
+read at startup from `LEASH_CAMERA_AIM_AUTH_TOKEN_FILE`, falling back to
+`LEASH_OPERATOR_AUTH_TOKEN_FILE`; no credential is included in status or logs.
+This path never creates or refreshes a pilot lease and never emits a wheel frame.
+Existing valid pilot-token callers remain supported. The camera bearer does not
+grant wheel authority or alter any wheel gate.
+
+Supply HTTP `Authorization: Bearer` privately, with a JSON body containing
+`pan_deg` and `tilt_deg`, optional `speed` and `accel`, and `approval` if required
+by the existing policy. MCP uses `invoke_capability` with
+`capability: camera_aim` and the same fields, supplying its authority as `token`.
+Pan must be finite and within -180..180 degrees; tilt within -30..90. Requests
+outside those bounds are rejected instead of clamped. Speed defaults to 8 and
+must be 1..40; acceleration defaults to 4 and must be 1..20, in the controller's
+units. Requests are serialized and must start at least 250 ms apart. These are
+bounded operating settings, not measurements of actual gimbal velocity.
+
+The only frame is the existing T133 command with `X`, `Y`, `SPD`, and `ACC`,
+queued through the existing single serial owner. The runtime estop check and
+the owner's estop latch both refuse camera commands while latched, including
+commands already queued when the owner observes the latch.
+
+Success retains `ok`, `pan_deg`, `tilt_deg`, `speed`, and `accel`, and adds
+`authority` (`camera-only-bearer` on this path) plus `serial_receipt` containing
+`controller_sequence`, `write_completed_at_ms`, and `physical_feedback: unknown`.
+It means the owner completed a serial write/flush, not that the requested physical
+pose was verified. No head command occurs merely by enabling this authority.
+
+`GET /camera/aim` retains the `gimbal` object. `gimbal.pose` remains null until a
+successful command; afterward it explicitly reports `source: last-commanded`
+and its write receipt. `gimbal.physical_pose_known` remains false. The new
+`gimbal.dedicated_authority` reports `enabled`, `token_configured`,
+`grants_wheel_authority: false`, request interval, last attempt time, and last
+write error. The limits/defaults are also published separately. A UI must not
+turn an unknown physical pose into an assumed centered gimbal.