Publish durable Session artifacts from hosted native outputs

CONTRIBUTING.md

@@ -424,6 +424,30 @@ large objects; live deletion does not erase WAL or historical backups. Schema
 rollback must not orphan existing source objects. Do not reuse product capability
 tables or introduce a second destination writer for file_id.
 
+Session Artifacts are immutable published output copies, separate from live
+workspace files and general source Files. A private output exporter must reuse
+the authorized workspace path boundary and stream bounded bytes. Require complete
+capture and confirmed helper/transport success before publication; valid archive
+syntax alone is insufficient. Never extract an output archive into Core's
+filesystem or hold the global execution lease through a large transfer. Keep
+publication ordered with Turn completion, and authorize stored reads independently
+of Environment availability so published outputs can survive its expiration.
+Capture bytes into private PostgreSQL large objects without a Session admission
+lock; after confirmed export, lock and recheck the live Turn before staging metadata.
+Before capture, seal native input under that lock using the private capture marker.
+Later messages reuse the existing Environment input reservation and await the next
+Turn; the public Turn stays in progress until publication settles. Directory reads
+during capture use an independent authorized read-only preparation, not the released
+native Run; they do not request model credentials or mutate the workspace. Cancellation
+retains the existing Turn/reservation semantics. Do not introduce a second queue.
+Publish metadata in the same transaction as Turn completion. Failed/cancelled Turns
+discard private objects, and Session deletion removes both private and published
+copies. Reuse the source-file snapshot reader pattern and common content response;
+artifact deletion does not alter workspace files. Hosted execution requires the
+Runtime's bounded output-export capability and exact read-only preparation binding;
+capability advertisement alone does not qualify an operator's deployment.
+Exporter component checks do not establish public Artifact compatibility.
+
 Local inline file delivery uses the same authenticated daemon connection and exact
 Environment/Session binding. The optional startup-owned `PARSAR_RUNTIME_WRITE_HELPER`
 and `PARSAR_RUNTIME_STAGING` enable only the bounded installer primitive; they do

apps/parsar-daemon/internal/cli/connect.go

@@ -372,12 +372,17 @@ func pumpConn(parentCtx context.Context, conn *transport.Conn, registry *agent.R
 	}()
 
 	conn.StartHeartbeats(parentCtx, boot.HeartbeatInterval(), func() proto.HeartbeatPayload {
+		kinds := registry.SupportedAgentKinds()
+		for i := range kinds {
+			caps := &kinds[i].Capabilities
+			caps.WorkspaceOutputExport = local.CanExport() && caps.LocalEnvironment && caps.WorkspaceReadPreparation
+		}
 		return proto.HeartbeatPayload{
 			Timestamp:           time.Now().Unix(),
 			ActiveRequests:      router.ActiveRuns(),
 			DaemonVersion:       Version,
 			ClaudeAvailable:     agentCLIs.ClaudeCode.Available, // legacy server compatibility
-			SupportedAgentKinds: registry.SupportedAgentKinds(),
+			SupportedAgentKinds: kinds,
 		}
 	}, obslog.Bg().With("component", "heartbeat"))
 

apps/parsar-daemon/internal/dispatch/preparation.go

@@ -73,7 +73,7 @@ func (r *Router) handleExecutionPrepare(ctx context.Context, env proto.Envelope)
 		r.mu.Unlock()
 		return ErrRouterClosed
 	}
-	if r.workspaceWrite != nil && !req.WorkspaceReadOnly {
+	if (r.workspaceWrite != nil || r.workspaceExport != nil) && !req.WorkspaceReadOnly {
 		r.mu.Unlock()
 		return r.rejectPreparation(env, "resource_unavailable")
 	}

apps/parsar-daemon/internal/dispatch/preparation_start.go

@@ -22,6 +22,10 @@ func (r *Router) handleExecutionStart(_ context.Context, env proto.Envelope) err
 		r.mu.Unlock()
 		return ErrRouterClosed
 	}
+	if r.workspaceExport != nil {
+		r.mu.Unlock()
+		return r.rejectPreparation(env, "resource_unavailable")
+	}
 	p := r.preparations[input.Handle]
 	if p == nil || p.requestID != env.ID {
 		r.mu.Unlock()

apps/parsar-daemon/internal/dispatch/prompt.go

@@ -74,7 +74,7 @@ func (r *Router) handlePromptRequest(callerCtx context.Context, env proto.Envelo
 		r.log.ErrorContext(callerCtx, "handlePromptRequest: router closed", "run_id", runID)
 		return ErrRouterClosed
 	}
-	if r.workspaceWrite != nil {
+	if r.workspaceWrite != nil || r.workspaceExport != nil {
 		r.mu.Unlock()
 		err := errors.New("local workspace has an unsettled write")
 		r.emitTerminalError(callerCtx, runID, err.Error())

apps/parsar-daemon/internal/dispatch/router.go

@@ -53,6 +53,7 @@ type Router struct {
 	preparationRequests map[string]*preparationState
 	preparationTimeout  time.Duration
 	workspaceWrite      *workspaceUpload
+	workspaceExport     *workspaceExport
 	workspaceReads      map[string]struct{}
 	localWorkspace      *localworkspace.Binding
 }
@@ -160,6 +161,8 @@ func (r *Router) Handle(ctx context.Context, env proto.Envelope) error {
 	ctx = adoptEnvelopeTrace(ctx, env)
 
 	switch env.Type {
+	case proto.TypeWorkspaceExport:
+		return r.handleWorkspaceExport(ctx, env)
 	case proto.TypeWorkspaceWrite:
 		return r.handleWorkspaceWrite(ctx, env)
 	case proto.TypeWorkspaceRead:

apps/parsar-daemon/internal/dispatch/workspace_export.go

@@ -0,0 +1,134 @@
+package dispatch
+
+import (
+	"context"
+	"errors"
+	"io"
+	"time"
+
+	"github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto"
+)
+
+type workspaceExport struct {
+	id       string
+	requests chan proto.WorkspaceExportPayload
+	cancel   context.CancelFunc
+}
+
+func (r *Router) handleWorkspaceExport(ctx context.Context, env proto.Envelope) error {
+	var request proto.WorkspaceExportPayload
+	if len(env.ID) == 0 || len(env.ID) > proto.WorkspaceReadMaxIDBytes || len(env.Payload) > proto.WorkspaceReadMaxRequestBytes || env.DecodePayload(&request) != nil || !proto.ValidWorkspaceExportRequest(request) {
+		return errors.New("dispatch: invalid workspace export request")
+	}
+	r.mu.Lock()
+	if r.closed {
+		r.mu.Unlock()
+		return ErrRouterClosed
+	}
+	u := r.workspaceExport
+	if request.Step != "begin" {
+		if u == nil || u.id != env.ID {
+			r.mu.Unlock()
+			return r.sendWorkspaceExport(ctx, env.ID, proto.WorkspaceExportResultPayload{Outcome: "rejected", ErrorCode: "resource_unavailable"})
+		}
+		if request.Step == "cancel" {
+			u.cancel()
+			r.mu.Unlock()
+			return nil
+		}
+		select {
+		case u.requests <- request:
+			r.mu.Unlock()
+			return nil
+		default:
+			u.cancel()
+			r.mu.Unlock()
+			return errors.New("dispatch: workspace export request already pending")
+		}
+	}
+	_, code := r.workspaceResourceLocked(proto.WorkspaceReadPayload{Handle: request.Handle, EnvironmentID: request.EnvironmentID})
+	p := r.preparations[request.Handle]
+	if u != nil || r.workspaceWrite != nil || !r.localWorkspace.CanExport() || code != "" || p == nil || !p.workspaceReadOnly {
+		r.mu.Unlock()
+		return r.sendWorkspaceExport(ctx, env.ID, proto.WorkspaceExportResultPayload{Outcome: "rejected", ErrorCode: "resource_unavailable"})
+	}
+	owner, stop := r.shutdownContext(p.ctx)
+	owner, cancel := context.WithTimeout(owner, 180*time.Second)
+	u = &workspaceExport{id: env.ID, requests: make(chan proto.WorkspaceExportPayload, 1), cancel: func() { cancel(); stop() }}
+	u.requests <- request
+	r.workspaceExport = u
+	r.shutdownWG.Add(1)
+	r.mu.Unlock()
+	go r.runWorkspaceExport(owner, u)
+	return nil
+}
+
+func (r *Router) runWorkspaceExport(ctx context.Context, u *workspaceExport) {
+	defer r.shutdownWG.Done()
+	reader, writer := io.Pipe()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		err := r.localWorkspace.ExportOutputs(ctx, writer)
+		_ = writer.CloseWithError(err)
+	}()
+	defer func() {
+		u.cancel()
+		_ = reader.Close()
+		<-done
+		r.mu.Lock()
+		if r.workspaceExport == u {
+			r.workspaceExport = nil
+		}
+		r.mu.Unlock()
+	}()
+	// Closing the reader unblocks a pending pipe read on cancellation or shutdown.
+	stopRead := context.AfterFunc(ctx, func() { _ = reader.CloseWithError(ctx.Err()) })
+	defer stopRead()
+	var offset int64
+	buffer := make([]byte, proto.WorkspaceExportChunkBytes)
+	for {
+		select {
+		case request := <-u.requests:
+			if request.Offset != offset {
+				_ = r.sendWorkspaceExport(ctx, u.id, proto.WorkspaceExportResultPayload{Outcome: "failed", Offset: offset, ErrorCode: "invalid_request"})
+				return
+			}
+		case <-ctx.Done():
+			return
+		}
+		n, err := reader.Read(buffer)
+		result := proto.WorkspaceExportResultPayload{Offset: offset}
+		switch {
+		case err == io.EOF:
+			result.Outcome = "completed"
+		case err != nil || n == 0 || int64(n) > proto.WorkspaceExportMaxBytes-offset:
+			result.Outcome, result.ErrorCode = "failed", "export_failed"
+		default:
+			result.Outcome, result.Data = "chunk", buffer[:n]
+			offset += int64(n)
+		}
+		if result.Outcome == "completed" {
+			// The next owner may start immediately after receiving completion.
+			<-done
+			r.mu.Lock()
+			if r.workspaceExport == u {
+				r.workspaceExport = nil
+			}
+			r.mu.Unlock()
+		}
+		if r.sendWorkspaceExport(ctx, u.id, result) != nil || result.Outcome != "chunk" {
+			return
+		}
+	}
+}
+
+func (r *Router) sendWorkspaceExport(ctx context.Context, id string, result proto.WorkspaceExportResultPayload) error {
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	env, err := proto.NewEnvelope(proto.TypeWorkspaceExportResult, id, result)
+	if err != nil {
+		return err
+	}
+	return r.sender.Send(ctx, env)
+}