feat(node): add execution-bound artifact reads

docs/releases/unreleased/ST-59-artifact-read.md

@@ -0,0 +1,2 @@
+- Node 新增绑定执行记录与视图的产物读取接口：逐段拒绝符号链接，拒绝硬链接和特殊文件，按块限制读取量，文件在读取期间变化时拒绝返回内容。
+- 这是产物读取迁移的底层准备；现有预览与知识库仍沿用原有路径，尚未完成全部迁移，也不代表运行隔离已经完成。

hub-rs/crates/steve-hub/src/wire/node_runtime.rs

@@ -753,6 +753,7 @@ impl NodeRuntime {
             // set in the bind's transaction, once its principal is recorded
             env: vec![],
             state: None,
+            artifact_view: None,
         };
         let fresh = self.tx(|tx| {
             self.session_owners_work_in(tx, &b.session)?;

hub-rs/crates/steve-node-api/src/discovery.rs

@@ -67,6 +67,8 @@ pub struct Capabilities {
 /// The Node writes a secret to a temporary file for a command ([`crate::SecretDelivery::File`]).
 /// Only the Hub's own machine offers it; a remote Node's claim is dropped.
 pub const FEATURE_SECRET_FILE: &str = "secret-file";
+/// Strict artifact reads bound to an accepted execution and its recorded view.
+pub const FEATURE_ARTIFACT_READ: &str = "artifact-read-v1";
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct HistoryRef {

hub-rs/crates/steve-node-api/src/ports.rs

@@ -85,6 +85,14 @@ pub trait WorkspaceExecutor: Send + Sync {
     /// Browse an allowed root (choosing a project directory).
     async fn list_root(&self, at: RootPath, page: PageRequest) -> ApiResult<Page<FileInfo>>;
     async fn read_file(&self, q: FileRead) -> ApiResult<ReadChunk>;
+    /// Strict, execution-scoped artifact read. Unsupported peers must never fall back to a
+    /// general filesystem read (which can follow links and does not bind an execution view).
+    async fn read_artifact(&self, _q: ArtifactRead) -> ApiResult<ReadChunk> {
+        Err(ApiError::new(
+            ErrorCode::Unsupported,
+            "execution-scoped artifact reads unsupported",
+        ))
+    }
     /// Scan a tree; read its pages under the scan's operation id.
     async fn scan(&self, c: Command<ScanSpec>) -> ApiResult<Receipt>;
     async fn scan_page(&self, scan: OperationId, page: PageRequest) -> ApiResult<ScanPage>;

hub-rs/crates/steve-node-api/src/run.rs

@@ -188,6 +188,10 @@ pub struct OpenSession {
     /// Node's, as before. An older Node ignores it.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub state: Option<HarnessState>,
+    /// The Hub's recorded view for artifact reads. Absent on older sessions: artifact reads
+    /// are unsupported until the Hub explicitly opens a session with a view.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub artifact_view: Option<Digest>,
 }
 
 /// A principal's own directory for one harness (path model §4.5.1): its configuration, logins,

hub-rs/crates/steve-node-api/src/testkit.rs

@@ -2392,6 +2392,7 @@ pub mod build {
             resume: None,
             env: Default::default(),
             state: Default::default(),
+            artifact_view: None,
         }
     }
 

hub-rs/crates/steve-node-api/src/workspace.rs

@@ -445,6 +445,23 @@ pub struct FileRead {
     pub max: u32,
 }
 
+/// A file produced by one accepted execution. Unlike a general [`FileRead`], this cannot name
+/// an arbitrary root: the Node takes the directory from the execution's durable artifact grant.
+/// The Hub authorizes the reader before calling this face. A digest is a scope binding, not a
+/// bearer token. Initially only session turns with an explicit artifact view are supported.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ArtifactRead {
+    pub execution: crate::common::ExecutionRef,
+    pub view_digest: Digest,
+    /// Plain relative names below the execution's workspace; no symlink is followed.
+    pub location: RelPath,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub expected: Option<Revision>,
+    pub offset: u64,
+    /// Bounded by the Node and transport even when the caller asks for more.
+    pub max: u32,
+}
+
 /// A directory tree to list for a knowledge source (or a project's description).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ScanSpec {

hub-rs/crates/steve-node-api/tests/contract.rs

@@ -86,6 +86,14 @@ fn types_survive_the_wire() {
         offset: 0,
         max: 10,
     });
+    roundtrip(&ArtifactRead {
+        execution: exec_turn("s_1", "t_1", "a_1"),
+        view_digest: Digest::sha256(b"view"),
+        location: RelPath::new("report.txt").unwrap(),
+        expected: Some(Revision::new("v1").unwrap()),
+        offset: 123,
+        max: 10,
+    });
 }
 
 #[test]

hub-rs/crates/steve-node-wire/src/methods.rs

@@ -103,6 +103,7 @@ rpc_table! {
     FS_LIST = "fs.list": HubToNode Control Read false;
     FS_LIST_ROOT = "fs.list_root": HubToNode Control Read false;
     FS_READ = "fs.read": HubToNode Bulk Blob false;
+    ARTIFACT_READ = "artifact.read": HubToNode Bulk Blob false;
     TREE_SCAN = "tree.scan": HubToNode Control Command true;
     SCAN_PAGE = "scan.page": HubToNode Control Read false;
     // Discovery

hub-rs/crates/steve-node-wire/src/node_server.rs

@@ -61,6 +61,7 @@ impl<N: Node + ?Sized + 'static> Handler for NodeServer<N> {
                 en(&n.list_root(a.at, a.page).await?)
             }
             FS_READ => en(&n.read_file(de(p)?).await?),
+            ARTIFACT_READ => en(&n.read_artifact(de(p)?).await?),
             TREE_SCAN => en(&n.scan(de(p)?).await?),
             SCAN_PAGE => {
                 let a: PagesArgs = de(p)?;

hub-rs/crates/steve-node-wire/src/remote_node.rs

@@ -261,6 +261,10 @@ impl WorkspaceExecutor for RemoteNode {
         q.max = q.max.min(crate::envelope::CHUNK_BYTES);
         self.go(FS_READ, &q, None, false).await
     }
+    async fn read_artifact(&self, mut q: ArtifactRead) -> ApiResult<ReadChunk> {
+        q.max = q.max.min(crate::envelope::CHUNK_BYTES);
+        self.go(ARTIFACT_READ, &q, None, false).await
+    }
     async fn scan(&self, c: Command<ScanSpec>) -> ApiResult<Receipt> {
         if c.body.limits != ScanLimits::default() {
             self.needs(2, "scan limits")?;

hub-rs/crates/steve-node-wire/tests/wire_faces.rs

@@ -137,6 +137,21 @@ async fn faces(n: Arc<dyn Node>) -> Vec<String> {
         })
         .await,
     ));
+    // A peer without this face returns unsupported; never reinterpret it as fs.read.
+    out.push(show(
+        &n.read_artifact(ArtifactRead {
+            execution: ExecutionRef::Job {
+                job: JobId::new("job-art").unwrap(),
+                attempt: AttemptId::new("attempt-art").unwrap(),
+            },
+            view_digest: Digest::sha256(b"artifact view"),
+            location: rel("d/b.md"),
+            expected: None,
+            offset: 0,
+            max: u32::MAX,
+        })
+        .await,
+    ));
     let spec = ScanSpec {
         root: at("notes", ""),
         include: vec![],
@@ -224,8 +239,9 @@ async fn the_faces_answer_over_the_wire_exactly_like_the_node() {
         assert_eq!(x, y, "step {i}");
     }
     assert!(a[0].starts_with("ok"), "{}", a[0]);
-    assert!(a[7].starts_with("ok"), "the push went through: {}", a[7]);
-    assert!(a[11].contains("invalid"), "a forged key: {}", a[11]);
+    assert!(a[3].contains("unsupported"), "old face refused: {}", a[3]);
+    assert!(a[8].starts_with("ok"), "the push went through: {}", a[8]);
+    assert!(a[12].contains("invalid"), "a forged key: {}", a[12]);
     assert_eq!(direct_secrets.taken(), hub_secrets.taken());
     assert_eq!(hub_secrets.taken().len(), 1);
     assert_eq!(direct.log(), fake.log());
@@ -335,6 +351,49 @@ impl Handler for Echo {
     }
 }
 
+struct ArtifactPeer(std::sync::Mutex<Option<ArtifactRead>>);
+
+#[async_trait]
+impl Handler for ArtifactPeer {
+    async fn handle(&self, method: &str, p: Value) -> ApiResult<Value> {
+        if method != ARTIFACT_READ {
+            return Err(unsupported(method));
+        }
+        let q: ArtifactRead = from_value(p)?;
+        let result = ReadChunk {
+            offset: q.offset,
+            bytes: b"artifact bytes".to_vec(),
+            eof: true,
+        };
+        *self.0.lock().unwrap() = Some(q);
+        Ok(serde_json::to_value(result).unwrap())
+    }
+}
+
+#[tokio::test]
+async fn artifact_reads_preserve_scope_and_bound_the_wire_chunk() {
+    let (h, n) = pair();
+    let hub = Conn::spawn(h.sink, h.stream, 1, Arc::new(Echo), opts("h"));
+    let peer = Arc::new(ArtifactPeer(Default::default()));
+    let _node = Conn::spawn(n.sink, n.stream, 1, peer.clone(), opts("n"));
+    let remote = steve_node_wire::RemoteNode::new(NodeId::new("n1").unwrap());
+    remote.attach(hub, false);
+    let mut q = ArtifactRead {
+        execution: exec_turn("s-art", "t-art", "a-art"),
+        view_digest: Digest::sha256(b"principal view"),
+        location: rel("report.txt"),
+        expected: Some(Revision::new("v-art").unwrap()),
+        offset: 123,
+        max: u32::MAX,
+    };
+    let result = remote.read_artifact(q.clone()).await.unwrap();
+    assert_eq!(result.bytes, b"artifact bytes");
+    assert_eq!(result.offset, 123);
+    q.max = steve_node_wire::envelope::CHUNK_BYTES;
+    assert_eq!(*peer.0.lock().unwrap(), Some(q));
+    assert_eq!(spec(ARTIFACT_READ).unwrap().lane, Lane::Bulk);
+}
+
 /// Both ends of a link that echoes (keep the second alive).
 fn echo_link() -> (Arc<Conn>, Arc<Conn>) {
     let (h, n) = pair();
@@ -564,7 +623,7 @@ impl Handler for OldPeer {
     async fn handle(&self, method: &str, p: Value) -> ApiResult<Value> {
         self.seen.lock().unwrap().push(method.to_string());
         match method {
-            HISTORY_LOCATE => Err(unsupported(method)),
+            HISTORY_LOCATE | ARTIFACT_READ => Err(unsupported(method)),
             GIT_INSPECT => {
                 // what it understands, and nothing else
                 let q: OldQuery = from_value(p)?;
@@ -701,6 +760,27 @@ async fn an_older_peer_is_never_sent_what_it_would_ignore() {
     };
     remote.scan(cmd("op_s2", default_scan)).await.unwrap();
     assert!(remote.probe(false).await.is_ok());
+    // A new method cannot be mistaken for an older method with ignored fields. A real old
+    // endpoint rejects it explicitly, and the remote never retries it through fs.read.
+    refused(
+        remote
+            .read_artifact(ArtifactRead {
+                execution: exec_turn("s-art", "t-art", "a-art"),
+                view_digest: Digest::sha256(b"view"),
+                location: rel("report.txt"),
+                expected: None,
+                offset: 0,
+                max: 100,
+            })
+            .await
+            .unwrap_err(),
+    );
+    assert_eq!(
+        peer.seen.lock().unwrap().last().map(String::as_str),
+        Some(ARTIFACT_READ)
+    );
+    assert!(!peer.seen.lock().unwrap().iter().any(|m| m == FS_READ));
+    assert!(remote.probe(false).await.is_ok(), "same link still usable");
     // a malformed payload is still invalid, not unsupported
     let e = hub
         .call(GIT_INSPECT, json!({"workspace": 1}), None)