diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md
index ea48434d1..c9a4caf0f 100644
--- a/crates/taskito-core/BINDING_CONTRACT.md
+++ b/crates/taskito-core/BINDING_CONTRACT.md
@@ -116,12 +116,14 @@ bytes of the section above, unchanged. `MAX_HEADER_BYTES` (64 KiB) and
 
 | Frame | Direction | Payload |
 |---|---|---|
-| `hello` | executor → scheduler | `{executor_id, sdk, version, tasks[], slots, protocol_version}` |
-| `hello_ack` | scheduler → executor | `{scheduler_id, protocol_version}` |
+| `hello` | executor → scheduler | `{executor_id, sdk, version, tasks[], slots, protocol_version, token?}` |
+| `hello_ack` | scheduler → executor | `{scheduler_id, protocol_version, capabilities[]}` |
 | `heartbeat` | executor → scheduler | `{free_slots}` |
-| `job` | scheduler → executor | `{id, task_name, payload_len, retry_count, max_retries, queue, timeout_ms, namespace}` + blob |
+| `job` | scheduler → executor | `{id, task_name, payload_len, retry_count, max_retries, queue, timeout_ms, namespace, disabled_middleware[], metadata}` + blob |
 | `cancel` | scheduler → executor | `{job_id}` |
 | `shutdown` | scheduler → executor | — |
+| `progress` | executor → scheduler | `{job_id, progress}` |
+| `task_log` | executor → scheduler | `{job_id, task_name, level, message, extra_len}` + blob |
 | `success` | executor → scheduler | `{job_id, result_len, task_name, wall_time_ns}` + blob |
 | `failure` | executor → scheduler | `{job_id, error, retry_count, max_retries, task_name, wall_time_ns, should_retry, timed_out}` |
 | `cancelled` | executor → scheduler | `{job_id, task_name, wall_time_ns}` |
@@ -132,9 +134,27 @@ Rules:
   is never silently downgraded. The scheduler sends `hello_ack` even when it is
   rejecting, so both ends can log both versions.
 - `result_len: null` means the task returned nothing; `0` means it returned an
-  empty value. They are distinct.
+  empty value. They are distinct. `extra_len` follows the same rule.
 - `should_retry` is the executor's decision — only it can see the exception. The
   core never inspects one.
+- **Optional behaviour is negotiated, not versioned.** `capabilities` lists what
+  the scheduler will do on an executor's behalf; `side_channel` means it applies
+  `progress` and `task_log` frames to storage. An executor sends neither frame
+  unless it was advertised. Adding one never bumps `protocol_version`, which
+  would force scheduler and executors to upgrade together.
+- **An unknown frame type is skipped, not fatal.** A reader that cannot name a
+  frame reads its declared payload length, discards that many bytes, logs once
+  and continues — the stream stays aligned, and a session keeps its in-flight
+  jobs. A frame type whose name *is* known but whose header will not parse stays
+  an error: that is a desync, not a newer peer.
+- **A new frame type must declare its payload length as `payload_len`.** It is
+  the only field a peer that predates the frame can find. `result_len` and
+  `extra_len` predate this rule and are read as equivalents.
+- `progress` and `task_log` are fire-and-forget: they carry no reply, never
+  settle a job, and a scheduler drops one naming a job the sender is not
+  running. `progress` is 0–100; a value outside that range is never written —
+  SDK boundaries clamp where they can, and the scheduler drops what still
+  reaches it rather than failing the job over a progress report.
 
 ## Task errors (structured, cross-SDK)
 When a task raises, the shell reports the failure as a **canonical JSON object**
diff --git a/crates/taskito-core/Cargo.toml b/crates/taskito-core/Cargo.toml
index 64dea84c5..1432fdac0 100644
--- a/crates/taskito-core/Cargo.toml
+++ b/crates/taskito-core/Cargo.toml
@@ -40,6 +40,10 @@ redis = { workspace = true, optional = true }
 
 [dev-dependencies]
 tempfile = "3"
+# Hand-rolled wire frames in the executor-attach tests: a frame type this build
+# has no variant for is exactly what the forward-compatibility path is about, so
+# the test has to be able to serialize one.
+serde = { workspace = true }
 # Raw connection access in the Redis contract tests, to set up race-window
 # state and assert on internal index keys. Only used by `#[cfg(feature = "redis")]`
 # tests; harmless when the feature is off.
diff --git a/crates/taskito-core/src/worker/executor.rs b/crates/taskito-core/src/worker/executor.rs
index 56a034ea6..0003465af 100644
--- a/crates/taskito-core/src/worker/executor.rs
+++ b/crates/taskito-core/src/worker/executor.rs
@@ -30,7 +30,7 @@
 //! # }
 //! ```
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, PoisonError};
 use std::thread::{self, JoinHandle};
@@ -41,8 +41,8 @@ use tokio::sync::mpsc::error::TrySendError;
 
 use super::auth::Secret;
 use super::protocol::{
-    ExecutorMessage, FrameReader, FrameWriter, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL,
-    PROTOCOL_VERSION,
+    ExecutorMessage, FrameReader, FrameWriter, Incoming, ProtocolError, SchedulerMessage,
+    CAP_SIDE_CHANNEL, PROTOCOL_VERSION,
 };
 use super::transport::{Connection, ReadHalf, Transport, WriteHalf};
 use super::WorkerDispatcher;
@@ -380,6 +380,17 @@ impl ExecutorSideChannel {
         self.shared.side_channel
     }
 
+    /// Task log lines shed since this executor attached, because they were
+    /// produced faster than they could be framed.
+    ///
+    /// Progress has no equivalent: it coalesces, so a backlog collapses instead
+    /// of losing anything. Exposed because "the log line I wrote is not in the
+    /// dashboard" is otherwise indistinguishable from a bug, and an SDK can
+    /// surface this as the backpressure signal it is.
+    pub fn dropped_task_logs(&self) -> u64 {
+        self.shared.dropped_logs.load(Ordering::Relaxed)
+    }
+
     /// Middleware the operator has disabled for a running job's task.
     ///
     /// Resolved by the scheduler at dispatch and carried on the job frame, so
@@ -806,19 +817,34 @@ fn spawn_reader(
     thread::Builder::new()
         .name("taskito-executor-reader".to_string())
         .spawn(move || {
+            // Once per type per session: a newer scheduler may send the frame we
+            // cannot read on every dispatch.
+            let mut reported_unknown: HashSet<String> = HashSet::new();
             loop {
-                match reader.read::<SchedulerMessage>() {
-                    Ok((SchedulerMessage::Shutdown, _)) => {
+                match reader.read_or_skip::<SchedulerMessage>() {
+                    Ok(Incoming::Known(SchedulerMessage::Shutdown, _)) => {
                         log::info!(
                             "[taskito] scheduler asked executor {} to shut down",
                             shared.executor_id
                         );
                         break;
                     }
-                    Ok((SchedulerMessage::Cancel { job_id }, _)) => {
+                    Ok(Incoming::Known(SchedulerMessage::Cancel { job_id }, _)) => {
                         dispatcher.notify_cancel(&job_id);
                     }
-                    Ok((frame, payload)) => accept_job(&shared, frame, payload),
+                    Ok(Incoming::Known(frame, payload)) => accept_job(&shared, frame, payload),
+                    // A frame from a scheduler newer than this executor. The
+                    // session is otherwise fine, and ending it would fail every
+                    // job in flight over a frame this build never needed.
+                    Ok(Incoming::Unknown { frame_type }) => {
+                        if reported_unknown.insert(frame_type.clone()) {
+                            log::warn!(
+                                "[taskito] scheduler sent a '{frame_type}' frame executor {} does \
+                                 not know; ignoring it (upgrade the executor to use it)",
+                                shared.executor_id
+                            );
+                        }
+                    }
                     Err(ProtocolError::Eof) => {
                         log::info!(
                             "[taskito] scheduler closed the connection to executor {}",
diff --git a/crates/taskito-core/src/worker/mod.rs b/crates/taskito-core/src/worker/mod.rs
index 246838a31..190c6ae4f 100644
--- a/crates/taskito-core/src/worker/mod.rs
+++ b/crates/taskito-core/src/worker/mod.rs
@@ -19,7 +19,8 @@ pub use executor::{
     ExecutorSideChannel,
 };
 pub use protocol::{
-    Dispatch, ExecutorMessage, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL, PROTOCOL_VERSION,
+    Dispatch, ExecutorMessage, Incoming, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL,
+    PROTOCOL_VERSION,
 };
 pub use registry::{TaskError, TaskHandler, TaskRegistry, TaskResult};
 pub use remote::{AttachError, AttachedExecutor, Capacity, RemoteConfig, RemoteDispatcher};
diff --git a/crates/taskito-core/src/worker/protocol.rs b/crates/taskito-core/src/worker/protocol.rs
index dd3427e99..c9bb6876a 100644
--- a/crates/taskito-core/src/worker/protocol.rs
+++ b/crates/taskito-core/src/worker/protocol.rs
@@ -12,6 +12,14 @@
 //! wire *are* the wire-envelope bytes of `BINDING_CONTRACT.md`. Headers stay
 //! JSON so every SDK can write an executor with its standard library alone.
 //! The same format serves a pipe (prefork's stdio children) and a socket.
+//!
+//! Because a header declares its own payload length, a peer can skip a frame
+//! type it has never heard of and stay aligned — see
+//! [`FrameReader::read_or_skip`], which is how a live session survives the far
+//! side being upgraded first. That only holds while the length stays findable
+//! without knowing the frame's shape: **a frame type added from here on must
+//! declare its payload length in a field named `payload_len`**. The two frames
+//! that predate the rule (`success`, `task_log`) are aliased into it.
 
 use std::io::{BufRead, BufWriter, Read, Write};
 
@@ -276,6 +284,13 @@ pub enum ExecutorMessage {
 pub trait Frame: Serialize + DeserializeOwned {
     /// Bytes of payload that follow this header; zero for frames carrying none.
     fn payload_len(&self) -> usize;
+
+    /// Whether `tag` names a frame type this build can parse.
+    ///
+    /// Tells a frame from a *newer* peer, which is safe to skip, from a corrupt
+    /// one of a type we do know — a real disagreement that must not be papered
+    /// over, because skipping it would silently lose a dispatch or a result.
+    fn is_known_type(tag: &str) -> bool;
 }
 
 impl Frame for SchedulerMessage {
@@ -285,6 +300,10 @@ impl Frame for SchedulerMessage {
             _ => 0,
         }
     }
+
+    fn is_known_type(tag: &str) -> bool {
+        matches!(tag, "hello_ack" | "job" | "cancel" | "shutdown")
+    }
 }
 
 impl Frame for ExecutorMessage {
@@ -295,6 +314,47 @@ impl Frame for ExecutorMessage {
             _ => 0,
         }
     }
+
+    fn is_known_type(tag: &str) -> bool {
+        matches!(
+            tag,
+            "hello" | "heartbeat" | "progress" | "task_log" | "success" | "failure" | "cancelled"
+        )
+    }
+}
+
+/// One frame off the wire, which the reader may not be able to name.
+///
+/// Returned by [`FrameReader::read_or_skip`], the tolerant read a live session
+/// uses. The handshake keeps the strict [`FrameReader::read`]: a peer that
+/// cannot say hello has nothing to offer, so there is nothing to be tolerant of.
+#[derive(Debug)]
+pub enum Incoming<F> {
+    /// A frame this build understands, with its payload.
+    Known(F, Vec<u8>),
+    /// A frame type this build does not know. Its payload has already been
+    /// consumed, so the stream is still aligned on a frame boundary.
+    Unknown {
+        /// The `type` tag the peer sent, for the log line that reports it.
+        frame_type: String,
+    },
+}
+
+/// The two things a frame this build cannot name must still yield: what it is,
+/// and how many payload bytes follow it.
+///
+/// The length is what makes an unknown frame skippable rather than fatal, so
+/// **any frame type added from now on must declare its payload length as
+/// `payload_len`** — a peer released before it has nothing else to go on. The
+/// two names that predate the rule, `result_len` (`success`) and `extra_len`
+/// (`task_log`), are aliased so a later frame modelled on either still skips
+/// cleanly rather than desyncing the wire.
+#[derive(Debug, Deserialize)]
+struct FramePreamble {
+    #[serde(rename = "type")]
+    frame_type: String,
+    #[serde(default, alias = "result_len", alias = "extra_len")]
+    payload_len: Option<usize>,
 }
 
 impl From<&Job> for SchedulerMessage {
@@ -631,8 +691,51 @@ impl<R: BufRead> FrameReader<R> {
     pub fn read<F: Frame>(&mut self) -> Result<(F, Vec<u8>), ProtocolError> {
         let header = self.read_header_line()?;
         let frame: F = serde_json::from_slice(&header)?;
+        let payload = self.read_payload(frame.payload_len())?;
+        Ok((frame, payload))
+    }
 
-        let len = frame.payload_len();
+    /// Read one frame, skipping past a type this build does not know.
+    ///
+    /// Scheduler and executor are released independently — that decoupling is
+    /// the whole reason an attach exists — so a frame type added on the far side
+    /// must degrade to "ignored". [`FrameReader::read`] would surface it as a
+    /// parse error, and the only thing a reader loop can do with one is drop the
+    /// connection, abandoning every job in flight to the reaper.
+    ///
+    /// Skipping is possible because a header declares its own payload length,
+    /// which the preamble reads without knowing the frame's shape. A *known*
+    /// type that fails to parse is still an error: that is a disagreement about
+    /// a frame we both claim to speak, not a newer peer.
+    pub fn read_or_skip<F: Frame>(&mut self) -> Result<Incoming<F>, ProtocolError> {
+        let header = self.read_header_line()?;
+        // Typed first, so a known frame pays nothing for the tolerance: the
+        // preamble is parsed only once this build has failed to name the frame.
+        let typed = match serde_json::from_slice::<F>(&header) {
+            Ok(frame) => {
+                let payload = self.read_payload(frame.payload_len())?;
+                return Ok(Incoming::Known(frame, payload));
+            }
+            Err(error) => error,
+        };
+
+        match serde_json::from_slice::<FramePreamble>(&header) {
+            // Not even a type tag: report the typed failure, which names what
+            // was expected. Nothing here says how far to skip, either.
+            Err(_) => Err(typed.into()),
+            Ok(preamble) if F::is_known_type(&preamble.frame_type) => Err(typed.into()),
+            Ok(preamble) => {
+                self.read_payload(preamble.payload_len.unwrap_or(0))?;
+                Ok(Incoming::Unknown {
+                    frame_type: preamble.frame_type,
+                })
+            }
+        }
+    }
+
+    /// Read exactly `len` payload bytes, capped so a corrupt length field cannot
+    /// allocate without bound.
+    fn read_payload(&mut self, len: usize) -> Result<Vec<u8>, ProtocolError> {
         if len > MAX_PAYLOAD_BYTES {
             return Err(ProtocolError::PayloadTooLarge { len });
         }
@@ -640,7 +743,7 @@ impl<R: BufRead> FrameReader<R> {
         if len > 0 {
             self.inner.read_exact(&mut payload)?;
         }
-        Ok((frame, payload))
+        Ok(payload)
     }
 
     /// Read through the header's newline, capped so a peer that never sends one
@@ -1267,4 +1370,137 @@ mod tests {
             Err(ProtocolError::Json(_))
         ));
     }
+
+    /// The `type` tag a frame actually serializes to.
+    fn wire_type<F: Frame>(frame: &F) -> String {
+        let value: serde_json::Value = serde_json::to_value(frame).expect("serialize");
+        value["type"].as_str().expect("a type tag").to_string()
+    }
+
+    #[test]
+    fn every_frame_reports_its_own_wire_type_as_known() {
+        // `is_known_type` is what separates "a peer newer than us" from "a frame
+        // we should have been able to read". A tag missing from it would make a
+        // real desync look like forward compatibility and skip a job's result.
+        for frame in [
+            SchedulerMessage::HelloAck {
+                scheduler_id: "s".into(),
+                protocol_version: PROTOCOL_VERSION,
+                capabilities: vec![],
+            },
+            SchedulerMessage::from(&sample_job(b"")),
+            SchedulerMessage::Cancel {
+                job_id: "job-1".into(),
+            },
+            SchedulerMessage::Shutdown,
+        ] {
+            let tag = wire_type(&frame);
+            assert!(
+                SchedulerMessage::is_known_type(&tag),
+                "'{tag}' serializes but is not listed as known"
+            );
+        }
+
+        for frame in [
+            hello_with_token(None),
+            ExecutorMessage::Heartbeat { free_slots: 1 },
+            ExecutorMessage::Progress {
+                job_id: "job-1".into(),
+                progress: 10,
+            },
+            ExecutorMessage::task_log("job-1", "t", "info", "m", None).0,
+            ExecutorMessage::Success {
+                job_id: "job-1".into(),
+                result_len: None,
+                task_name: "t".into(),
+                wall_time_ns: 1,
+            },
+            ExecutorMessage::Failure {
+                job_id: "job-1".into(),
+                error: "e".into(),
+                retry_count: 0,
+                max_retries: 3,
+                task_name: "t".into(),
+                wall_time_ns: 1,
+                should_retry: true,
+                timed_out: false,
+            },
+            ExecutorMessage::Cancelled {
+                job_id: "job-1".into(),
+                task_name: "t".into(),
+                wall_time_ns: 1,
+            },
+        ] {
+            let tag = wire_type(&frame);
+            assert!(
+                ExecutorMessage::is_known_type(&tag),
+                "'{tag}' serializes but is not listed as known"
+            );
+        }
+    }
+
+    #[test]
+    fn an_unknown_frame_is_skipped_and_the_next_one_still_reads() {
+        // The forward-compatibility contract: a peer released after this build
+        // can send frames it does not know, and the stream stays aligned.
+        let mut buf = Vec::new();
+        buf.extend_from_slice(b"{\"type\":\"telemetry\",\"payload_len\":5}\nabcde");
+        FrameWriter::new(&mut buf)
+            .write_header(&ExecutorMessage::Heartbeat { free_slots: 2 })
+            .expect("write");
+
+        let mut reader = FrameReader::new(buf.as_slice());
+        match reader.read_or_skip::<ExecutorMessage>().expect("skip") {
+            Incoming::Unknown { frame_type } => assert_eq!(frame_type, "telemetry"),
+            Incoming::Known(frame, _) => panic!("unexpectedly parsed {frame:?}"),
+        }
+        assert!(matches!(
+            reader.read_or_skip::<ExecutorMessage>().expect("read"),
+            Incoming::Known(ExecutorMessage::Heartbeat { free_slots: 2 }, _)
+        ));
+    }
+
+    #[test]
+    fn an_unknown_frame_declaring_a_legacy_length_field_is_still_skipped() {
+        // `success` and `task_log` predate the `payload_len` rule, so a later
+        // frame modelled on either would name its blob `result_len`/`extra_len`.
+        // Honouring both is what keeps that copy-paste from desyncing the wire.
+        let mut buf = Vec::new();
+        buf.extend_from_slice(b"{\"type\":\"partial\",\"extra_len\":2}\nhi");
+        FrameWriter::new(&mut buf)
+            .write_header(&ExecutorMessage::Heartbeat { free_slots: 1 })
+            .expect("write");
+
+        let mut reader = FrameReader::new(buf.as_slice());
+        assert!(matches!(
+            reader.read_or_skip::<ExecutorMessage>().expect("skip"),
+            Incoming::Unknown { .. }
+        ));
+        assert!(matches!(
+            reader.read_or_skip::<ExecutorMessage>().expect("read"),
+            Incoming::Known(ExecutorMessage::Heartbeat { free_slots: 1 }, _)
+        ));
+    }
+
+    #[test]
+    fn a_known_frame_type_that_will_not_parse_is_still_an_error() {
+        // Skipping this would silently drop a dispatch. Only an *unrecognised*
+        // type is forward compatibility; a broken `job` is a disagreement.
+        let buf = b"{\"type\":\"job\",\"id\":\"job-1\"}\n";
+        assert!(matches!(
+            FrameReader::new(&buf[..]).read_or_skip::<SchedulerMessage>(),
+            Err(ProtocolError::Json(_))
+        ));
+    }
+
+    #[test]
+    fn a_header_without_a_type_tag_is_an_error_rather_than_a_skip() {
+        // Nothing here says how far to skip, so pretending otherwise would
+        // desync the stream on the very next read.
+        let buf = b"{\"payload_len\":4}\n";
+        assert!(matches!(
+            FrameReader::new(&buf[..]).read_or_skip::<ExecutorMessage>(),
+            Err(ProtocolError::Json(_))
+        ));
+    }
 }
diff --git a/crates/taskito-core/src/worker/remote.rs b/crates/taskito-core/src/worker/remote.rs
index f5f791927..485247dab 100644
--- a/crates/taskito-core/src/worker/remote.rs
+++ b/crates/taskito-core/src/worker/remote.rs
@@ -16,8 +16,8 @@ use tokio::sync::Notify;
 
 use super::auth::Secret;
 use super::protocol::{
-    ExecutorMessage, FrameReader, FrameWriter, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL,
-    PROTOCOL_VERSION,
+    ExecutorMessage, FrameReader, FrameWriter, Incoming, ProtocolError, SchedulerMessage,
+    CAP_SIDE_CHANNEL, PROTOCOL_VERSION,
 };
 use super::side_channel::SideChannel;
 use super::transport::{Connection, ReadHalf, Transport, WriteHalf};
@@ -926,14 +926,35 @@ impl Shared {
         thread::Builder::new()
             .name(format!("taskito-executor-{}", executor.id))
             .spawn(move || {
+                // Reported once per type per connection: a newer executor may
+                // send the frame we cannot read on every job, and a warning per
+                // frame would bury the one that explains the upgrade.
+                let mut reported_unknown: HashSet<String> = HashSet::new();
                 loop {
-                    match reader.read::<ExecutorMessage>() {
-                        Ok((message, payload)) => {
+                    match reader.read_or_skip::<ExecutorMessage>() {
+                        Ok(Incoming::Known(message, payload)) => {
                             executor
                                 .last_seen_ms
                                 .store(self.elapsed_ms(), Ordering::Relaxed);
                             self.handle_frame(&executor, message, payload);
                         }
+                        // A frame from an executor newer than this scheduler.
+                        // Skipped, not fatal: the peer is plainly alive, and
+                        // dropping the attach would abandon its in-flight jobs
+                        // to the reaper over a frame nobody needed.
+                        Ok(Incoming::Unknown { frame_type }) => {
+                            executor
+                                .last_seen_ms
+                                .store(self.elapsed_ms(), Ordering::Relaxed);
+                            if reported_unknown.insert(frame_type.clone()) {
+                                log::warn!(
+                                    "[taskito] executor {} sent a '{frame_type}' frame this \
+                                     scheduler does not know; ignoring it (upgrade the scheduler \
+                                     to use it)",
+                                    executor.id
+                                );
+                            }
+                        }
                         Err(ProtocolError::Eof) => {
                             log::info!("[taskito] executor {} disconnected", executor.id);
                             break;
diff --git a/crates/taskito-core/tests/rust/executor_tests.rs b/crates/taskito-core/tests/rust/executor_tests.rs
index eaa0c32cb..b6f9e131b 100644
--- a/crates/taskito-core/tests/rust/executor_tests.rs
+++ b/crates/taskito-core/tests/rust/executor_tests.rs
@@ -5,13 +5,15 @@
 //! side could only prove it agrees with itself.
 
 use std::collections::HashMap;
+use std::io::{self, Write};
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Condvar, Mutex};
 use std::thread;
 use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
 use crossbeam_channel::{Receiver, Sender};
+use serde::{Deserialize, Serialize};
 
 use taskito_core::job::{Job, JobStatus};
 use taskito_core::scheduler::JobResult;
@@ -20,11 +22,13 @@ use taskito_core::worker::executor::{
     ExecutorClient, ExecutorConfig, ExecutorError, ExecutorHandle,
 };
 use taskito_core::worker::protocol::{
-    ExecutorMessage, FrameReader, FrameWriter, ProtocolError, SchedulerMessage, CAP_SIDE_CHANNEL,
-    PROTOCOL_VERSION,
+    ExecutorMessage, Frame, FrameReader, FrameWriter, ProtocolError, SchedulerMessage,
+    CAP_SIDE_CHANNEL, PROTOCOL_VERSION,
 };
 use taskito_core::worker::remote::{RemoteConfig, RemoteDispatcher};
-use taskito_core::worker::transport::{MemoryTransport, ReadHalf, Transport, WriteHalf};
+use taskito_core::worker::transport::{
+    Connection, MemoryTransport, ReadHalf, Transport, WriteHalf,
+};
 use taskito_core::worker::WorkerDispatcher;
 
 const SETTLE: Duration = Duration::from_secs(5);
@@ -240,6 +244,128 @@ fn dial(
     connected
 }
 
+/// A frame from a peer released after this build: a type tag with no variant
+/// behind it, plus the payload length that makes it skippable.
+#[derive(Debug, Serialize, Deserialize)]
+struct FutureFrame {
+    #[serde(rename = "type")]
+    frame_type: String,
+    payload_len: usize,
+}
+
+impl FutureFrame {
+    fn new(frame_type: &str, payload: &[u8]) -> Self {
+        Self {
+            frame_type: frame_type.to_string(),
+            payload_len: payload.len(),
+        }
+    }
+}
+
+impl Frame for FutureFrame {
+    fn payload_len(&self) -> usize {
+        self.payload_len
+    }
+
+    /// Never: the whole point of this frame is that the peer cannot name it.
+    fn is_known_type(_tag: &str) -> bool {
+        false
+    }
+}
+
+/// A latch a test can shut to hold the executor's writer off the wire.
+///
+/// The result loop frames everything under one lock, so parking a write is how
+/// the outbound queue behind it is filled deterministically — racing the drain
+/// would only fill it by luck.
+#[derive(Default)]
+struct Gate {
+    open: Mutex<bool>,
+    changed: Condvar,
+}
+
+impl Gate {
+    fn opened() -> Arc<Self> {
+        Arc::new(Self {
+            open: Mutex::new(true),
+            changed: Condvar::new(),
+        })
+    }
+
+    fn close(&self) {
+        *self.open.lock().expect("gate") = false;
+    }
+
+    fn release(&self) {
+        *self.open.lock().expect("gate") = true;
+        self.changed.notify_all();
+    }
+
+    fn wait(&self) {
+        let mut open = self.open.lock().expect("gate");
+        while !*open {
+            open = self.changed.wait(open).expect("gate");
+        }
+    }
+}
+
+/// A [`MemoryTransport`] whose write half parks while its [`Gate`] is shut.
+struct StallingTransport {
+    inner: Box<MemoryTransport>,
+    gate: Arc<Gate>,
+}
+
+impl Transport for StallingTransport {
+    fn split(self: Box<Self>) -> io::Result<(ReadHalf, WriteHalf, Connection)> {
+        let (read, write, connection) = self.inner.split()?;
+        Ok((
+            read,
+            Box::new(StallingWriter {
+                inner: write,
+                gate: self.gate,
+            }),
+            connection,
+        ))
+    }
+
+    fn peer(&self) -> String {
+        self.inner.peer()
+    }
+}
+
+struct StallingWriter {
+    inner: WriteHalf,
+    gate: Arc<Gate>,
+}
+
+impl Write for StallingWriter {
+    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
+        self.gate.wait();
+        self.inner.write(data)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+/// How a test wants the executor's outbound side-channel wired.
+struct SideChannelTuning {
+    /// Queued operations allowed before the oldest task logs are shed.
+    capacity: usize,
+    /// Holds the executor's writer, so the queue behind it can be filled.
+    gate: Option<Arc<Gate>>,
+}
+
+impl Default for SideChannelTuning {
+    fn default() -> Self {
+        Self {
+            capacity: ExecutorConfig::new("test", "0.0.0").side_channel_capacity,
+            gate: None,
+        }
+    }
+}
+
 /// The scheduler end of the wire, hand-driven.
 ///
 /// [`RemoteDispatcher`] is the right peer for most of these tests, but it is
@@ -263,8 +389,44 @@ impl FakeScheduler {
         tasks: &[&str],
         slots: u32,
         capabilities: Vec<String>,
+    ) -> (Self, ExecutorHandle, Arc<TestPool>) {
+        Self::attach_tuned(tasks, slots, capabilities, SideChannelTuning::default())
+    }
+
+    /// Handshake with the outbound queue sized to `capacity` and the executor's
+    /// writer behind a gate, so a test can hold frames off the wire and fill it.
+    fn attach_stalled(
+        tasks: &[&str],
+        slots: u32,
+        capacity: usize,
+    ) -> (Self, ExecutorHandle, Arc<TestPool>, Arc<Gate>) {
+        let gate = Gate::opened();
+        let (scheduler, handle, pool) = Self::attach_tuned(
+            tasks,
+            slots,
+            vec![CAP_SIDE_CHANNEL.to_string()],
+            SideChannelTuning {
+                capacity,
+                gate: Some(gate.clone()),
+            },
+        );
+        (scheduler, handle, pool, gate)
+    }
+
+    fn attach_tuned(
+        tasks: &[&str],
+        slots: u32,
+        capabilities: Vec<String>,
+        tuning: SideChannelTuning,
     ) -> (Self, ExecutorHandle, Arc<TestPool>) {
         let (scheduler_end, executor_end) = MemoryTransport::pair();
+        let executor_end: Box<dyn Transport> = match tuning.gate {
+            Some(gate) => Box::new(StallingTransport {
+                inner: Box::new(executor_end),
+                gate,
+            }),
+            None => Box::new(executor_end),
+        };
 
         // `connect` blocks on the ack, so the scheduler side runs concurrently.
         let accepting = thread::spawn(move || {
@@ -291,13 +453,14 @@ impl FakeScheduler {
         });
 
         let client = ExecutorClient::connect(
-            Box::new(executor_end),
+            executor_end,
             ExecutorConfig {
                 executor_id: "exec-1".to_string(),
                 tasks: tasks.iter().map(|task| (*task).to_string()).collect(),
                 slots,
                 heartbeat_interval: Duration::from_millis(50),
                 shutdown_drain: Duration::from_secs(2),
+                side_channel_capacity: tuning.capacity,
                 ..ExecutorConfig::new("test", "0.0.0")
             },
         )
@@ -313,6 +476,13 @@ impl FakeScheduler {
         self.send_job_with(id, task_name, payload, Vec::new());
     }
 
+    /// Write a frame type no executor built today can name.
+    fn send_future_frame(&mut self, frame_type: &str, payload: &[u8]) {
+        self.writer
+            .write(&FutureFrame::new(frame_type, payload), payload)
+            .expect("send a future frame");
+    }
+
     fn send_job_with(
         &mut self,
         id: &str,
@@ -372,6 +542,74 @@ impl FakeScheduler {
     }
 }
 
+/// The executor end of the wire, hand-driven.
+///
+/// The mirror of [`FakeScheduler`], and needed for the same reason:
+/// [`ExecutorClient`] only ever writes frames this build knows, so proving a
+/// scheduler survives one it does not takes a peer that writes anything.
+struct FakeExecutor {
+    reader: FrameReader<ReadHalf>,
+    writer: FrameWriter<WriteHalf>,
+}
+
+impl FakeExecutor {
+    /// Handshake with a real [`RemoteDispatcher`], leaving both ends live.
+    fn attach(dispatcher: &RemoteDispatcher, tasks: &[&str], slots: u32) -> Self {
+        let (scheduler_end, executor_end) = MemoryTransport::pair();
+        let accepting = {
+            let dispatcher = dispatcher.clone();
+            thread::spawn(move || dispatcher.attach(Box::new(scheduler_end)))
+        };
+
+        let (read, write, _connection) = Box::new(executor_end).split().expect("split");
+        let mut executor = Self {
+            reader: FrameReader::new(read),
+            writer: FrameWriter::new(write),
+        };
+        executor
+            .writer
+            .write_header(&ExecutorMessage::Hello {
+                executor_id: "exec-fake".to_string(),
+                sdk: "test".to_string(),
+                version: "0.0.0".to_string(),
+                tasks: tasks.iter().map(|task| (*task).to_string()).collect(),
+                slots,
+                protocol_version: PROTOCOL_VERSION,
+                token: None,
+            })
+            .expect("send hello");
+        match executor.reader.read::<SchedulerMessage>().expect("ack").0 {
+            SchedulerMessage::HelloAck { .. } => {}
+            other => panic!("expected hello_ack, got {other:?}"),
+        }
+        accepting.join().expect("attach thread").expect("attach");
+        executor
+    }
+
+    /// Write a frame type no scheduler built today can name.
+    fn send_future_frame(&mut self, frame_type: &str, payload: &[u8]) {
+        self.writer
+            .write(&FutureFrame::new(frame_type, payload), payload)
+            .expect("send a future frame");
+    }
+
+    /// Read the next dispatch, answering it with a success.
+    fn run_next_job(&mut self) {
+        let job_id = match self.reader.read::<SchedulerMessage>().expect("read").0 {
+            SchedulerMessage::Job { id, .. } => id,
+            other => panic!("expected a job, got {other:?}"),
+        };
+        self.writer
+            .write_header(&ExecutorMessage::Success {
+                job_id,
+                result_len: None,
+                task_name: "resize".to_string(),
+                wall_time_ns: 1,
+            })
+            .expect("send success");
+    }
+}
+
 fn make_job(id: &str, task_name: &str, payload: &[u8]) -> Job {
     Job {
         id: id.to_string(),
@@ -1151,6 +1389,124 @@ fn a_flood_of_progress_neither_blocks_the_task_nor_grows_without_bound() {
     handle.shutdown();
 }
 
+#[test]
+fn a_full_outbound_queue_coalesces_progress_and_sheds_the_oldest_logs() {
+    // The two halves of the backpressure contract, in the one state that
+    // exercises both: a queue that is actually full. Progress is
+    // idempotent-latest so a backlog collapses to one value; log lines are data
+    // and cannot, so the only bounded answers are "park the task" or "drop".
+    //
+    // The writer is held shut rather than raced. With nothing draining, every
+    // push past the capacity *has* to shed and every progress report *has* to
+    // land on the same map entry — which is what makes either assertable at all
+    // instead of a coin flip against the drain.
+    const CAPACITY: usize = 4;
+    const FLOOD: usize = 500;
+
+    let (mut scheduler, handle, _pool, gate) =
+        FakeScheduler::attach_stalled(&["resize"], 1, CAPACITY);
+    let side_channel = handle.side_channel();
+
+    gate.close();
+    let flooding = Instant::now();
+    for line in 0..FLOOD {
+        side_channel.report_progress("job-1", (line % 101) as i32);
+        side_channel.write_task_log("job-1", "resize", "info", &format!("line-{line}"), None);
+    }
+    assert!(
+        flooding.elapsed() < SETTLE,
+        "a reporting loop must never park the task on the scheduler"
+    );
+
+    // The result loop can hold at most one line beyond the queue — the one it
+    // is parked mid-write on — so everything past that had to be shed.
+    let dropped = side_channel.dropped_task_logs();
+    let unavoidable = (FLOOD - CAPACITY - 1) as u64;
+    assert!(
+        dropped >= unavoidable,
+        "expected at least {unavoidable} shed lines, counted {dropped}"
+    );
+
+    gate.release();
+
+    // Drop-oldest, not drop-newest: under a flood the lines nearest the present
+    // are the ones worth keeping, so the last one written must still arrive.
+    let last_line = format!("line-{}", FLOOD - 1);
+    let newest_percent = ((FLOOD - 1) % 101) as i32;
+    let deadline = Instant::now() + SETTLE;
+    let mut newest_log = None;
+    let mut progress = Vec::new();
+    while newest_log.as_deref() != Some(last_line.as_str())
+        || progress.last() != Some(&newest_percent)
+    {
+        assert!(
+            Instant::now() < deadline,
+            "the queue never drained: last log {newest_log:?}, progress {progress:?}"
+        );
+        match scheduler.next_frame().0 {
+            ExecutorMessage::TaskLog { message, .. } => newest_log = Some(message),
+            ExecutorMessage::Progress {
+                progress: value, ..
+            } => progress.push(value),
+            _ => {}
+        }
+    }
+
+    // At most two: the flush already parked mid-write when the gate shut, plus
+    // the one that drains everything reported behind it. Never one per report.
+    assert!(
+        progress.len() <= 2,
+        "{FLOOD} reports must coalesce, got {progress:?}"
+    );
+
+    handle.shutdown();
+}
+
+#[test]
+fn a_frame_from_a_newer_scheduler_is_ignored_rather_than_ending_the_session() {
+    // An executor released before the scheduler must not treat a frame it has
+    // no variant for as a fatal desync: ending the session there would fail
+    // every job in flight over a frame this build never needed.
+    let (mut scheduler, handle, _pool) = FakeScheduler::attach(&["resize"], 1);
+
+    scheduler.send_future_frame("prefetch", b"payload-it-cannot-read");
+
+    // A job behind the unknown frame is the proof: it can only be read once the
+    // skip consumed exactly the right number of bytes.
+    scheduler.send_job("job-1", "resize", b"payload");
+    assert!(matches!(
+        scheduler.expect_result(),
+        ExecutorMessage::Success { job_id, .. } if job_id == "job-1"
+    ));
+
+    handle.shutdown();
+}
+
+#[test]
+fn a_frame_from_a_newer_executor_is_ignored_rather_than_abandoning_the_attach() {
+    // The same tolerance on the scheduler's reader. Dropping the attach here
+    // would deregister a live executor and leave its in-flight jobs for the
+    // reaper — a slow, visible failure caused by a frame nobody needed.
+    let dispatcher = scheduler(None);
+    let mut executor = FakeExecutor::attach(&dispatcher, &["resize"], 1);
+
+    executor.send_future_frame("telemetry", b"payload-it-cannot-read");
+
+    with_running(&dispatcher, |jobs, results| {
+        jobs.blocking_send(make_job("job-1", "resize", b"in"))
+            .expect("send job");
+        executor.run_next_job();
+        assert!(matches!(expect_result(results), JobResult::Success { .. }));
+        // Asserted while the dispatcher is still running: its own shutdown
+        // deregisters every executor, so afterwards this proves nothing.
+        assert_eq!(
+            dispatcher.executors().len(),
+            1,
+            "the executor must still be attached"
+        );
+    });
+}
+
 #[test]
 fn a_dispatched_toggle_list_is_readable_until_the_job_reports() {
     let (mut scheduler, handle, pool) = FakeScheduler::attach(&["resize"], 1);
diff --git a/docs/content/docs/shared/guides/operations/executor.mdx b/docs/content/docs/shared/guides/operations/executor.mdx
index 816320e5f..15958fdf1 100644
--- a/docs/content/docs/shared/guides/operations/executor.mdx
+++ b/docs/content/docs/shared/guides/operations/executor.mdx
@@ -139,6 +139,7 @@ def resize(path: str) -> str:
 queue.task("resize", async (path: string) => {
   const job = currentJob();
   job?.setProgress(50);
+  job?.log("halfway");
   job?.publish({ stage: "halfway" });
   return "done";
 });
@@ -189,6 +190,13 @@ or task-log frames, and those calls become no-ops (logged once). Everything else
 — dispatch, results, retries, cancellation — is unaffected. The executor logs
 which capabilities it negotiated at attach.
 
+Skew in the other direction is covered too. A frame type the receiving side has
+never heard of is logged once and skipped rather than treated as a corrupt
+stream: a header declares its own payload length, so the reader stays aligned on
+the next frame. That holds both ways, so neither an older scheduler nor an older
+executor drops a live session — and its in-flight jobs — over a frame it did not
+need.
+
 ## Trade-offs
 
 | Cost | Mitigation |
diff --git a/sdks/node/src/context.ts b/sdks/node/src/context.ts
index 5dfabeb85..cadc017df 100644
--- a/sdks/node/src/context.ts
+++ b/sdks/node/src/context.ts
@@ -1,4 +1,5 @@
 import { AsyncLocalStorage } from "node:async_hooks";
+import type { TaskLogLevel } from "./types";
 
 /**
  * Ambient context available to a running task via {@link currentJob}. Mirrors
@@ -11,6 +12,14 @@ export interface JobContext {
   readonly signal: AbortSignal;
   /** Report progress (0–100) for observability. */
   setProgress(progress: number): void;
+  /**
+   * Write a structured log line against this job, visible in the dashboard and
+   * through `queue.taskLogs(jobId)`.
+   *
+   * `extra` must be JSON-serializable; a value that is not is stored as its
+   * string form rather than failing the task.
+   */
+  log(message: string, level?: TaskLogWriteLevel, extra?: unknown): void;
   /**
    * Publish a partial result, consumable live via `queue.stream(jobId)`. The
    * value must be JSON-serializable. Use to stream progress from a long-running
@@ -19,6 +28,15 @@ export interface JobContext {
   publish(value: unknown): void;
 }
 
+/**
+ * Levels {@link JobContext.log} accepts.
+ *
+ * `result` is excluded: a published partial is a task log at that level, and
+ * {@link JobContext.publish} is how one is written. Letting `log` forge one
+ * would put an unencoded message where `queue.stream` expects a payload.
+ */
+export type TaskLogWriteLevel = Exclude<TaskLogLevel, "result">;
+
 const store = new AsyncLocalStorage<JobContext>();
 
 /** The context of the task running on this async stack, or `undefined`. */
diff --git a/sdks/node/src/detached.ts b/sdks/node/src/detached.ts
index 520a4a2ad..3b498c440 100644
--- a/sdks/node/src/detached.ts
+++ b/sdks/node/src/detached.ts
@@ -15,6 +15,60 @@ export function isDetached(): boolean {
   return process.env[DETACHED_ENV] === "1";
 }
 
+/**
+ * Where an executor's storage-shaped writes go instead of a database.
+ *
+ * Implemented over the attached executor handle, which frames each call to the
+ * scheduler — the side that still holds the connection. Both methods are
+ * fire-and-forget: a task reporting progress must not be able to fail, or
+ * block, because of what is happening at the far end.
+ */
+export interface ExecutorSink {
+  updateProgress(jobId: string, progress: number): void;
+  writeTaskLog(
+    jobId: string,
+    taskName: string,
+    level: string,
+    message: string,
+    extra?: string,
+  ): void;
+}
+
+/** The sink a single detached stand-in routes through, once one has attached. */
+interface SinkHolder {
+  current?: ExecutorSink;
+}
+
+/**
+ * Each detached stand-in's sink, keyed by the stand-in itself.
+ *
+ * Per queue rather than per process: two executors can attach from one process,
+ * and a single module-level slot gave the last attach the first executor's
+ * writes, then silenced whichever was still running when either one stopped.
+ */
+const sinks = new WeakMap<object, SinkHolder>();
+
+/** Route this detached queue's progress and task logs through `next`. @internal */
+export function installSink(queue: NativeQueue, next: ExecutorSink): void {
+  const holder = sinks.get(queue);
+  if (holder) {
+    holder.current = next;
+  }
+}
+
+/**
+ * Stop routing `installed`'s writes, so they degrade to a warning again.
+ *
+ * A no-op once something else holds the slot: an executor shutting down must
+ * not disconnect the one that replaced it. @internal
+ */
+export function clearSink(queue: NativeQueue, installed: ExecutorSink): void {
+  const holder = sinks.get(queue);
+  if (holder?.current === installed) {
+    holder.current = undefined;
+  }
+}
+
 /** An executor was asked for something only a database could answer. */
 export class DetachedStorageError extends Error {
   constructor(operation: string) {
@@ -48,17 +102,36 @@ const RUNTIME_PROBES = new Set([
  * The native queue's job-scoped conveniences, degraded.
  *
  * Reads answer empty, because that is exactly what a queue with no such row
- * returns and callers already handle it. Progress and task logs are dropped
- * with one warning rather than throwing: a task that only wanted to report
- * progress must not fail because it happens to be running detached.
+ * returns and callers already handle it. Progress and task logs are forwarded
+ * to the installed {@link ExecutorSink}: the executor has no storage, but the
+ * scheduler does, and it applies them on this process's behalf. Without a sink
+ * — an app imported outside `taskito executor`, or a scheduler that advertised
+ * no side-channel — they degrade to one warning rather than throwing, because a
+ * task that only wanted to report progress must not fail for running detached.
  */
-function degraded(warnOnce: (what: string) => void): Record<string, unknown> {
+function degraded(warnOnce: (what: string) => void, holder: SinkHolder): Record<string, unknown> {
   return {
-    updateProgress(): void {
-      warnOnce("setProgress");
+    updateProgress(jobId: string, progress: number): void {
+      const sink = holder.current;
+      if (!sink) {
+        warnOnce("setProgress");
+        return;
+      }
+      sink.updateProgress(jobId, progress);
     },
-    writeTaskLog(): void {
-      warnOnce("publish");
+    writeTaskLog(
+      jobId: string,
+      taskName: string,
+      level: string,
+      message: string,
+      extra?: string,
+    ): void {
+      const sink = holder.current;
+      if (!sink) {
+        warnOnce("log/publish");
+        return;
+      }
+      sink.writeTaskLog(jobId, taskName, level, message, extra);
     },
     // A cancel reaches an executor as a protocol frame; the executor overrides
     // this check with its own native state (see `Executor.start`).
@@ -101,13 +174,14 @@ export function createDetachedNative(): NativeQueue {
       warned.add(what);
       log.warn(
         () =>
-          `${what} is unavailable on an attached executor, which has no storage; ignoring. ` +
-          "Run an in-process worker if you need it.",
+          `${what} is unavailable on an attached executor with no side-channel to the ` +
+          "scheduler; ignoring. Run an in-process worker if you need it.",
       );
     }
   };
 
-  const supported = degraded(warnOnce);
+  const holder: SinkHolder = {};
+  const supported = degraded(warnOnce, holder);
   const proxy = new Proxy(supported, {
     get(target, property): unknown {
       if (typeof property !== "string") {
@@ -135,5 +209,9 @@ export function createDetachedNative(): NativeQueue {
   // The stand-in answers the calls a running task makes and throws on the rest,
   // so a union type would force every storage call site to handle a case only
   // an executor ever sees.
-  return proxy as unknown as NativeQueue;
+  const native = proxy as unknown as NativeQueue;
+  // Keyed by the stand-in the executor is handed, which is this one: that is how
+  // an attach finds the queue whose writes it is answering for.
+  sinks.set(native, holder);
+  return native;
 }
diff --git a/sdks/node/src/executor.ts b/sdks/node/src/executor.ts
index 742e076a7..a9dad4703 100644
--- a/sdks/node/src/executor.ts
+++ b/sdks/node/src/executor.ts
@@ -1,3 +1,4 @@
+import { clearSink, type ExecutorSink, installSink } from "./detached";
 import type { Emitter } from "./events";
 import type { Middleware } from "./middleware";
 import {
@@ -69,6 +70,9 @@ export class Executor {
 
   private constructor(
     private readonly native: NativeExecutor,
+    /** The detached stand-in this attach answers for, and whose sink it holds. */
+    private readonly queue: NativeQueue,
+    private readonly sink: ExecutorSink,
     private readonly resources: ResourceRuntime,
     private readonly emitter: Emitter,
     private readonly onStopped?: () => void,
@@ -119,6 +123,12 @@ export class Executor {
       resources,
       queue,
       isCancelled: (jobId) => attached?.isCancelRequested(jobId) ?? false,
+      // Overridden rather than left to `queue`, which is only the detached
+      // stand-in when this process is a `taskito executor`. An executor started
+      // from a process that *does* have storage would otherwise write progress
+      // into its own database, where the job it names does not exist — the row
+      // belongs to the scheduler. `queue`'s own route to the scheduler is the
+      // sink installed below, for app code that calls the queue directly.
       setProgress: (jobId, progress) => attached?.reportProgress(jobId, progress),
       writeTaskLog: (jobId, taskName, level, message, extra) =>
         attached?.writeTaskLog(jobId, taskName, level, message, extra),
@@ -143,6 +153,22 @@ export class Executor {
     });
 
     attached = native;
+    const sink: ExecutorSink = {
+      updateProgress: (jobId, progress) => native.reportProgress(jobId, progress),
+      writeTaskLog: (jobId, taskName, level, message, extra) =>
+        native.writeTaskLog(jobId, taskName, level, message, extra),
+    };
+    // Before `markAttached`, so the first invocation through the gate already
+    // has somewhere to report to. Scoped to `queue`, so a second executor in
+    // this process routes its own writes and neither steals the other's.
+    installSink(queue, sink);
+    if (!native.supportsSideChannel()) {
+      log.warn(
+        () =>
+          `scheduler ${native.schedulerId} applies no progress or task logs on this executor's ` +
+          "behalf; they will be dropped. Upgrade the scheduler to keep them.",
+      );
+    }
     markAttached();
     try {
       // Only lease the resource runtime once the attach actually succeeded, so a
@@ -159,7 +185,7 @@ export class Executor {
       throw error;
     }
 
-    return new Executor(native, resources, emitter, onStopped);
+    return new Executor(native, queue, sink, resources, emitter, onStopped);
   }
 
   /** Identity the scheduler announced when it accepted this attach. */
@@ -205,6 +231,10 @@ export class Executor {
     try {
       await this.native.shutdown();
     } finally {
+      // The sink frames to a session that is over; leaving it installed would
+      // turn a late report into a silent no-op instead of the warning it is.
+      // Only this attach's own — another executor may have taken the slot since.
+      clearSink(this.queue, this.sink);
       try {
         await this.resources.teardownWorker();
       } catch (error) {
diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts
index 952705b98..81a7b31da 100644
--- a/sdks/node/src/index.ts
+++ b/sdks/node/src/index.ts
@@ -11,7 +11,7 @@ export {
   WorkerProcessManager,
 } from "./autoscale";
 export { Batcher, type BatcherOptions } from "./batching";
-export { currentJob, type JobContext } from "./context";
+export { currentJob, type JobContext, type TaskLogWriteLevel } from "./context";
 export {
   AuthStore,
   bootstrapAdminFromEnv,
diff --git a/sdks/node/src/task-callback.ts b/sdks/node/src/task-callback.ts
index 4cef111f7..6dae79942 100644
--- a/sdks/node/src/task-callback.ts
+++ b/sdks/node/src/task-callback.ts
@@ -111,8 +111,18 @@ export function createTaskCallback(
       jobId: invocation.id,
       signal: controller.signal,
       setProgress: (progress) => setProgress(invocation.id, progress),
+      log: (message, level = "info", extra) =>
+        writeTaskLog(
+          invocation.id,
+          invocation.taskName,
+          level,
+          message,
+          extra === undefined ? undefined : encodeExtra(extra),
+        ),
+      // A published partial is a task log at level `result`, which is what lets
+      // `queue.stream` pick it out of ordinary logs.
       publish: (value) =>
-        writeTaskLog(invocation.id, invocation.taskName, "result", "", JSON.stringify(value)),
+        writeTaskLog(invocation.id, invocation.taskName, "result", "", encodeExtra(value)),
     };
     const poller = setInterval(() => {
       try {
@@ -183,6 +193,21 @@ export function createTaskCallback(
   };
 }
 
+/**
+ * Encode a structured `extra` blob for a task log.
+ *
+ * Falls back to the value's string form rather than throwing: a circular
+ * reference in a log line must not fail the task that wrote it, and a
+ * best-effort rendering is more use than a lost entry.
+ */
+function encodeExtra(value: unknown): string {
+  try {
+    return JSON.stringify(value) ?? String(value);
+  } catch {
+    return String(value);
+  }
+}
+
 /** Whether a task's `retryOn` predicate accepts this failure. */
 export function isRetryable(task: RegisteredTask, error: unknown): boolean {
   const predicate = task.options?.retryOn;
diff --git a/sdks/node/test/worker/detachedSink.test.ts b/sdks/node/test/worker/detachedSink.test.ts
new file mode 100644
index 000000000..d15fa9517
--- /dev/null
+++ b/sdks/node/test/worker/detachedSink.test.ts
@@ -0,0 +1,92 @@
+/**
+ * Sink routing belongs to a detached queue, not to the process.
+ *
+ * Two executors can attach from one Node process. While the sink was a single
+ * module-level slot, the second attach took the first one's writes, and either
+ * one stopping silenced whichever was still running.
+ */
+
+import { expect, it } from "vitest";
+import {
+  clearSink,
+  createDetachedNative,
+  type ExecutorSink,
+  installSink,
+} from "../../src/detached";
+
+interface Recorder extends ExecutorSink {
+  readonly progress: Array<[string, number]>;
+  readonly logs: string[];
+}
+
+function recorder(): Recorder {
+  const progress: Array<[string, number]> = [];
+  const logs: string[] = [];
+  return {
+    progress,
+    logs,
+    updateProgress: (jobId, value) => {
+      progress.push([jobId, value]);
+    },
+    writeTaskLog: (jobId, _taskName, _level, message) => {
+      logs.push(`${jobId}:${message}`);
+    },
+  };
+}
+
+it("routes each detached queue's writes to its own sink", () => {
+  const first = createDetachedNative();
+  const second = createDetachedNative();
+  const one = recorder();
+  const two = recorder();
+
+  installSink(first, one);
+  installSink(second, two);
+
+  first.updateProgress("job-1", 10);
+  second.updateProgress("job-2", 20);
+  first.writeTaskLog("job-1", "task", "info", "from one");
+
+  expect(one.progress).toEqual([["job-1", 10]]);
+  expect(two.progress).toEqual([["job-2", 20]]);
+  expect(one.logs).toEqual(["job-1:from one"]);
+  expect(two.logs).toEqual([]);
+});
+
+it("keeps one queue routing when another executor stops", () => {
+  const first = createDetachedNative();
+  const second = createDetachedNative();
+  const one = recorder();
+  const two = recorder();
+  installSink(first, one);
+  installSink(second, two);
+
+  clearSink(first, one);
+
+  first.updateProgress("job-1", 10);
+  second.updateProgress("job-2", 20);
+
+  expect(one.progress).toEqual([]);
+  expect(two.progress).toEqual([["job-2", 20]]);
+});
+
+it("ignores a clear from an executor that has already been replaced", () => {
+  const queue = createDetachedNative();
+  const gone = recorder();
+  const live = recorder();
+  installSink(queue, gone);
+  installSink(queue, live);
+
+  clearSink(queue, gone);
+  queue.updateProgress("job-1", 30);
+
+  expect(live.progress).toEqual([["job-1", 30]]);
+  expect(gone.progress).toEqual([]);
+});
+
+it("degrades rather than throwing when no executor has attached", () => {
+  const queue = createDetachedNative();
+
+  expect(() => queue.updateProgress("job-1", 10)).not.toThrow();
+  expect(() => queue.writeTaskLog("job-1", "task", "info", "dropped")).not.toThrow();
+});
diff --git a/sdks/node/test/worker/executorAttach.test.ts b/sdks/node/test/worker/executorAttach.test.ts
index 892f24d02..d8d617780 100644
--- a/sdks/node/test/worker/executorAttach.test.ts
+++ b/sdks/node/test/worker/executorAttach.test.ts
@@ -285,6 +285,29 @@ async function reportingHandler(): Promise<string> {
   return "reported";
 }
 
+/** The queue's job-scoped writes, which on an executor have no storage behind them. */
+interface JobScopedWrites {
+  updateProgress(jobId: string, progress: number): void;
+  writeTaskLog(
+    jobId: string,
+    taskName: string,
+    level: string,
+    message: string,
+    extra?: string,
+  ): void;
+}
+
+/**
+ * Reach the queue's native handle.
+ *
+ * Private on purpose, and the seam under test on purpose: this is what a task
+ * uses when it goes through the queue rather than `currentJob()`, and on an
+ * executor it has to reach the scheduler just the same.
+ */
+function nativeOf(queue: Queue): JobScopedWrites {
+  return (queue as unknown as { native: JobScopedWrites }).native;
+}
+
 /** Encode a call the way the enqueue path does. */
 function payloadFor(queue: Queue, args: unknown[]): Buffer {
   // biome-ignore lint/complexity/useLiteralKeys: reaching the internal serializer
@@ -644,6 +667,74 @@ it("sends progress and logs to a scheduler that advertised the side-channel", as
   }
 });
 
+it("sends the queue's own progress and log writes to the scheduler", async () => {
+  // The other seam: a task that reaches the queue directly rather than through
+  // `currentJob()`. On an executor that shim has no storage behind it, and
+  // warning-and-dropping there would make the same call silently poorer than
+  // the context one.
+  scheduler = await FakeScheduler.listen();
+  scheduler.capabilities = ["side_channel"];
+  process.env[DETACHED_ENV] = "1";
+  try {
+    const queue = new Queue({ backend: "postgres", dsn: "postgres://x:y@127.0.0.1:1/absent" });
+    queue.task("reports", () => {
+      const native = nativeOf(queue);
+      native.updateProgress("job-1", 25);
+      native.writeTaskLog("job-1", "reports", "warning", "from the queue", '{"via":"queue"}');
+      return "reported";
+    });
+
+    executor = await queue.runExecutor({ attach: `127.0.0.1:${scheduler.port}` });
+    await scheduler.attached();
+    scheduler.sendJob("job-1", "reports", payloadFor(queue, []));
+
+    const { result, sideChannel } = await scheduler.collectUntilResult();
+    expect(result.header.type).toBe("success");
+
+    const progress = sideChannel
+      .filter((frame) => frame.header.type === "progress")
+      .map((frame) => frame.header.progress);
+    expect(progress).toContain(25);
+
+    const logged = sideChannel.find((frame) => frame.header.level === "warning");
+    expect(logged?.header.message).toBe("from the queue");
+    expect(logged?.header.task_name).toBe("reports");
+    expect(JSON.parse(logged?.payload.toString() ?? "")).toEqual({ via: "queue" });
+  } finally {
+    delete process.env[DETACHED_ENV];
+  }
+});
+
+it("writes a structured log line from the job context", async () => {
+  // `publish` was the only route to a task log before this; a plain log line
+  // had nowhere to go on an executor at all.
+  scheduler = await FakeScheduler.listen();
+  scheduler.capabilities = ["side_channel"];
+  process.env[DETACHED_ENV] = "1";
+  try {
+    const queue = new Queue({ backend: "postgres", dsn: "postgres://x:y@127.0.0.1:1/absent" });
+    queue.task("logs", () => {
+      currentJob()?.log("halfway", "warning", { step: 2 });
+      return "logged";
+    });
+
+    executor = await queue.runExecutor({ attach: `127.0.0.1:${scheduler.port}` });
+    await scheduler.attached();
+    scheduler.sendJob("job-1", "logs", payloadFor(queue, []));
+
+    const { result, sideChannel } = await scheduler.collectUntilResult();
+    expect(result.header.type).toBe("success");
+
+    const logged = sideChannel.find((frame) => frame.header.type === "task_log");
+    expect(logged?.header.level).toBe("warning");
+    expect(logged?.header.message).toBe("halfway");
+    expect(logged?.header.job_id).toBe("job-1");
+    expect(JSON.parse(logged?.payload.toString() ?? "")).toEqual({ step: 2 });
+  } finally {
+    delete process.env[DETACHED_ENV];
+  }
+});
+
 it("skips a middleware the dispatch says is disabled", async () => {
   // A dashboard toggle has to reach a process that cannot read settings, so it
   // rides the job frame instead.