[feat] core: publish execution artifacts through commands

apps/api/src/opendirector_api/creation.py

@@ -56,6 +56,8 @@ async def execute(project_id: UUID, value: Command, x_opendirector_session: UUID
         if value.projectId != project_id:
             raise ProjectError('命令不属于当前工程。')
         session = str(x_opendirector_session) if x_opendirector_session else None
+        if value.type == 'artifact.publish':
+            raise ProjectError('产物发布需要服务端执行上下文，不能通过路径直接访问本机文件。', 403)
         if value.type == 'job.checkpoint':
             raise ProjectError('任务执行事实只能由后台写入。', 403)
         if value.type in {'generation.submit', 'job.cancel', 'job.recover'}:

packages/core/src/opendirector/application/creation_artifacts.py

@@ -0,0 +1,66 @@
+"""Publish execution files through the shared asset staging and commit services."""
+from contextlib import ExitStack
+import os
+from pathlib import Path
+import stat
+
+from opendirector.domain.creation.project import ProjectSnapshot
+from opendirector.domain.projects import ProjectError
+from opendirector.infrastructure.creation_files import CHUNK_BYTES
+from opendirector.infrastructure.creation_uploads import asset_name
+
+
+def publish_artifact(assets, command, *, execution_directory: Path, session, provenance, source='agent'):
+    """The host supplies the execution directory; command payloads cannot choose it."""
+    project_id, request = command.projectId, command.payload
+    projects = assets.projects
+    replay = projects.replay(command)
+    if replay is not None:
+        return replay
+    with projects.local.transaction() as db:
+        projects._recover()
+        directory, _, records = projects._load(db, project_id)
+        projects._lease(db, project_id, directory, session)
+        snapshot = records[0].snapshot
+        for reference in request.sources:
+            assets._reference(snapshot, reference)
+    if os.open not in os.supports_dir_fd or not hasattr(os, 'O_NOFOLLOW'):
+        raise ProjectError('当前平台不支持安全的执行产物文件读取。', 501)
+    work = execution_directory.resolve()
+    candidate = Path(request.path)
+    candidate = candidate if candidate.is_absolute() else work / candidate
+    # Normalize dot segments without following symlinks; the descriptor walk rejects links.
+    candidate = Path(os.path.abspath(candidate))
+    if not candidate.is_relative_to(work) or candidate == work:
+        raise ProjectError('只能发布本轮执行目录中的文件。', 403)
+    name = asset_name(request.name)
+    key = command.commandId
+    with ExitStack() as stack:
+        descriptor = os.open(work, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        stack.callback(os.close, descriptor)
+        parts = candidate.relative_to(work).parts
+        for part in parts[:-1]:
+            descriptor = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor)
+            stack.callback(os.close, descriptor)
+        file_descriptor = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=descriptor)
+        stack.callback(os.close, file_descriptor)
+        facts = os.fstat(file_descriptor)
+        if not stat.S_ISREG(facts.st_mode) or facts.st_size == 0:
+            raise ProjectError('发布产物必须是非空普通文件。')
+        stream = stack.enter_context(os.fdopen(os.dup(file_descriptor), 'rb'))
+        writer = assets.uploads.begin(project_id, key, session, name, expected_kind=request.expectedKind)
+        try:
+            while chunk := stream.read(CHUNK_BYTES):
+                writer.write(chunk)
+            after = os.fstat(stream.fileno())
+            if (facts.st_size, facts.st_mtime_ns, facts.st_ctime_ns) != (after.st_size, after.st_mtime_ns, after.st_ctime_ns):
+                raise ProjectError('发布期间文件发生变化，请等待文件写入完成后重试。', 409)
+            writer.finish()
+        finally:
+            writer.abort()
+        def attach(before, asset):
+            for reference in request.sources:
+                assets._reference(before, reference)
+            return ProjectSnapshot.model_validate({**before.model_dump(), 'assets': [*before.assets, asset]})
+        return assets.register(command, session=session, source=source, provenance=provenance.model_copy(update={'sources': request.sources}),
+            apply=attach, upload_handle=key)

packages/core/src/opendirector/application/creation_assets.py

@@ -69,7 +69,7 @@ def _reference(cls, snapshot, reference):
             raise ContractError('reference_invalid', 'Asset reference differs from its content', asset.id)
         return asset
 
-    def register(self, command, *, session=None, source='ui', provenance=None, apply=None):
+    def register(self, command, *, session=None, source='ui', provenance=None, apply=None, upload_handle=None):
         """Trusted services may supply provenance and atomically update jobs/editors.
 
         HTTP never accepts these extra arguments. Generated/edited outputs
@@ -79,17 +79,17 @@ def register(self, command, *, session=None, source='ui', provenance=None, apply
         if replay is not None:
             return replay
         payload = command.payload
-        if command.type != 'asset.import':
+        if command.type != 'asset.import' and not (command.type == 'artifact.publish' and upload_handle is not None):
             raise ProjectError('文件登记需要素材导入命令。')
-        handle = payload.uploadHandle
+        handle = payload.uploadHandle if command.type == 'asset.import' else upload_handle
         with self.uploads.lock(command.projectId, handle):
             _, owner, record = self.uploads.ready(command.projectId, handle, session)
             asset = Asset(id=upload_asset_id(command.projectId, handle), projectId=command.projectId,
                 kind=record.kind, name=asset_name(payload.name), fingerprint=record.fingerprint,
                 path=f'assets/{upload_asset_id(command.projectId, handle)}', metadata=record.metadata,
                 provenance=provenance or AssetProvenance(source='upload'))
             def register(before):
-                if payload.expectedKind != record.kind:
+                if payload.expectedKind is not None and payload.expectedKind != record.kind:
                     raise ContractError('invalid_input', 'Uploaded media kind differs from the command')
                 existing = next((item for item in before.assets if item.id == asset.id), None)
                 if existing:

packages/core/src/opendirector/domain/creation/commands.py

@@ -25,6 +25,14 @@ class AssetImport(Contract):
     expectedKind: MediaKind
 
 
+class ArtifactPublish(Contract):
+    """Publish a file from a server-owned execution context without generating media."""
+    path: str = Field(min_length=1, max_length=4096)
+    name: str = Field(min_length=1, max_length=255)
+    expectedKind: MediaKind | None = None
+    sources: tuple[AssetRef, ...] = Field(default=(), max_length=100)
+
+
 class AssetWriteText(Contract):
     name: str = Field(min_length=1, max_length=255)
     text: str = Field(max_length=1024 * 1024)
@@ -277,6 +285,7 @@ def scope_and_versions(self):
     ('ProjectTrashCommand', 'project.trash', EmptyPayload),
     ('ProjectRestoreCommand', 'project.restore', EmptyPayload),
     ('AssetImportCommand', 'asset.import', AssetImport),
+    ('ArtifactPublishCommand', 'artifact.publish', ArtifactPublish),
     ('AssetWriteTextCommand', 'asset.writeText', AssetWriteText),
     ('AssetMetadataCommand', 'asset.updateMetadata', AssetMetadataUpdate),
     ('AssetDeleteCommand', 'asset.delete', AssetDelete),