fix: recover run runtime state and logs

2026-08-07 12:01:26 +08:00
parent 3818fa0344
commit b789925ae5
23 changed files with 297 additions and 78 deletions
@@ -137,6 +137,7 @@ func (h *coreHandlers) register(mux *http.ServeMux) {
 	mux.HandleFunc("/api/v1/run/jobs/cancel", h.requireRunSignature(h.runJobCancelPoll))
 	mux.HandleFunc("/api/v1/run/jobs/reconcile", h.requireRunSignature(h.runJobReconcile))
 	mux.HandleFunc("/api/v1/run/logs/batches", h.requireRunSignature(h.runLogBatchIngest))
+	mux.HandleFunc("/api/v1/run/logs/progress", h.requireRunSignature(h.runLogStreamProgress))
 	mux.HandleFunc("/api/v1/run/artifacts/open", h.requireRunSignature(h.runArtifactOpen))
 	mux.HandleFunc("/api/v1/run/artifacts/chunks", h.requireRunSignature(h.runArtifactChunkUpload))
 	mux.HandleFunc("/api/v1/run/artifacts/status", h.requireRunSignature(h.runArtifactStatus))
@@ -2037,6 +2038,24 @@ func (h *coreHandlers) runLogBatchIngest(w http.ResponseWriter, r *http.Request)
 	writeJSON(w, http.StatusOK, dto.LogBatchIngestFromDomain(result))
 }

+func (h *coreHandlers) runLogStreamProgress(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeMethodNotAllowed(w, http.MethodPost)
+		return
+	}
+	request, err := decodeJSON[dto.RunLogStreamProgressRequest](r)
+	if err != nil {
+		writeDecodeError(w, err)
+		return
+	}
+	result, err := h.core.GetRunLogStreamProgress(request.ToDomain())
+	if err != nil {
+		writeServiceError(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, dto.RunLogStreamProgressFromDomain(result))
+}
+
 // runArtifactOpen godoc
 // @Summary Open run artifact upload transfer
 // @Description Lets a registered run endpoint open a resumable upload transfer for a scoped artifact owner.
@@ -62,6 +62,9 @@ type RunLifecycleReport struct {
 	Progress         RunJobProgressReport
 	Message          string
 	ErrorCode        string
+	ManagedProcessID string
+	ObservationSeq   uint64
+	ObservedAt       time.Time
 	ExecutionResult  JobExecutionResult
 }

@@ -48,6 +48,24 @@ type LogStreamCursorResult struct {
 	LatestSeq   uint64
 }

+// RunLogStreamProgress is a signed Run-only request for sequence recovery.
+// It deliberately exposes no log body or host-local information.
+type RunLogStreamProgress struct {
+	RunEndpointID    string
+	SessionToken     string
+	ServerInstanceID string
+	LogStreamID      string
+}
+
+type RunLogStreamProgressResult struct {
+	Accepted         bool
+	RunEndpointID    string
+	ServerInstanceID string
+	LogStreamID      string
+	LatestSeq        uint64
+	ServerTime       time.Time
+}
+
 type LogStreamEvent struct {
 	ServerInstanceID string
 	Stream           LogStream
@@ -777,19 +777,22 @@ type ServerInstance struct {
 	// DeploymentTargetID identifies an optional operator-selected deployment
 	// target for post-creation deployment operations. It never selects a
 	// distribution builder or replaces the server's generated Run endpoint.
-	DeploymentTargetID string
-	RunEndpointID      string
-	Name               string
-	OwnerUserID        string
-	AdminUserIDs       []string
-	State              ServerInstanceState
-	ConfigVersion      int
-	ConfigKey          string
-	ConfigContent      string
-	ConfigChecksum     string
-	ConfigUpdatedAt    time.Time
-	CreatedAt          time.Time
-	UpdatedAt          time.Time
+	DeploymentTargetID      string
+	RunEndpointID           string
+	Name                    string
+	OwnerUserID             string
+	AdminUserIDs            []string
+	State                   ServerInstanceState
+	LifecycleProcessID      string
+	LifecycleObservationSeq uint64
+	LifecycleObservedAt     time.Time
+	ConfigVersion           int
+	ConfigKey               string
+	ConfigContent           string
+	ConfigChecksum          string
+	ConfigUpdatedAt         time.Time
+	CreatedAt               time.Time
+	UpdatedAt               time.Time
 	// Deployment stores operator-supplied deployment inputs. Its protected path
 	// and command values are never included in normal platform read projections.
 	Deployment           ServerDeploymentDefinition
@@ -66,6 +66,9 @@ type RunLifecycleReportRequest struct {
 	Progress         JobProgressBody           `json:"progress"`
 	Message          string                    `json:"message,omitempty"`
 	ErrorCode        string                    `json:"errorCode,omitempty"`
+	ManagedProcessID string                    `json:"managedProcessId,omitempty"`
+	ObservationSeq   uint64                    `json:"observationSeq,omitempty"`
+	ObservedAt       time.Time                 `json:"observedAt,omitempty"`
 	ExecutionResult  RunJobExecutionResultBody `json:"executionResult,omitempty"`
 }

@@ -122,6 +125,9 @@ func (request RunLifecycleReportRequest) ToDomain() domain.RunLifecycleReport {
 		Progress:         progressReportToDomain(request.Progress),
 		Message:          request.Message,
 		ErrorCode:        request.ErrorCode,
+		ManagedProcessID: request.ManagedProcessID,
+		ObservationSeq:   request.ObservationSeq,
+		ObservedAt:       request.ObservedAt,
 		ExecutionResult:  domain.JobExecutionResult{Kind: request.ExecutionResult.Kind, ProcessState: request.ExecutionResult.ProcessState, ExitClassification: request.ExecutionResult.ExitClassification, ExitCode: request.ExecutionResult.ExitCode, Version: request.ExecutionResult.Version, Checksum: request.ExecutionResult.Checksum, SizeBytes: request.ExecutionResult.SizeBytes, AuditSummary: request.ExecutionResult.AuditSummary, Content: request.ExecutionResult.Content, ServerDeploymentEvidence: serverDeploymentEvidenceToDomain(request.ExecutionResult.ServerDeploymentEvidence), DeploymentReceipt: deploymentReceiptToDomain(request.ExecutionResult.DeploymentReceipt)},
 	}
 }
@@ -53,6 +53,22 @@ type LogStreamCursorResponse struct {
 	LatestSeq   uint64         `json:"latestSeq"`
 }

+type RunLogStreamProgressRequest struct {
+	RunEndpointID    string `json:"runEndpointId"`
+	SessionToken     string `json:"sessionToken"`
+	ServerInstanceID string `json:"serverInstanceId"`
+	LogStreamID      string `json:"logStreamId"`
+}
+
+type RunLogStreamProgressResponse struct {
+	Accepted         bool      `json:"accepted"`
+	RunEndpointID    string    `json:"runEndpointId"`
+	ServerInstanceID string    `json:"serverInstanceId"`
+	LogStreamID      string    `json:"logStreamId"`
+	LatestSeq        uint64    `json:"latestSeq"`
+	ServerTime       time.Time `json:"serverTime"`
+}
+
 type LogStreamEventResponse struct {
 	ServerInstanceID string                 `json:"serverInstanceId"`
 	StreamID         string                 `json:"streamId"`
@@ -92,6 +108,10 @@ func (request LogStreamCursorRequest) ToDomain() domain.LogStreamCursorQuery {
 	}
 }

+func (request RunLogStreamProgressRequest) ToDomain() domain.RunLogStreamProgress {
+	return domain.RunLogStreamProgress{RunEndpointID: request.RunEndpointID, SessionToken: request.SessionToken, ServerInstanceID: request.ServerInstanceID, LogStreamID: request.LogStreamID}
+}
+
 func LogBatchIngestFromDomain(result domain.LogBatchIngestResult) LogBatchIngestResponse {
 	return LogBatchIngestResponse{
 		Accepted:     result.Accepted,
@@ -114,6 +134,10 @@ func LogStreamCursorFromDomain(result domain.LogStreamCursorResult) LogStreamCur
 	}
 }

+func RunLogStreamProgressFromDomain(result domain.RunLogStreamProgressResult) RunLogStreamProgressResponse {
+	return RunLogStreamProgressResponse{Accepted: result.Accepted, RunEndpointID: result.RunEndpointID, ServerInstanceID: result.ServerInstanceID, LogStreamID: result.LogStreamID, LatestSeq: result.LatestSeq, ServerTime: result.ServerTime}
+}
+
 func LogStreamEventFromDomain(event domain.LogStreamEvent) LogStreamEventResponse {
 	event = domain.CopyLogStreamEvent(event)
 	return LogStreamEventResponse{
@@ -587,6 +587,7 @@ type ServerInstanceResponse struct {
 	OwnerUserID          string                         `json:"ownerUserId,omitempty"`
 	AdminUserIDs         []string                       `json:"adminUserIds"`
 	State                domain.ServerInstanceState     `json:"state"`
+	LifecycleObservedAt  *time.Time                     `json:"lifecycleObservedAt,omitempty"`
 	ConfigVersion        int                            `json:"configVersion"`
 	ConfigKey            string                         `json:"configKey,omitempty"`
 	ConfigChecksum       string                         `json:"configChecksum,omitempty"`
@@ -1640,6 +1641,7 @@ func ServerInstanceFromDomain(instance domain.ServerInstance) ServerInstanceResp
 		OwnerUserID:          instance.OwnerUserID,
 		AdminUserIDs:         adminUserIDs,
 		State:                instance.State,
+		LifecycleObservedAt:  optionalTime(instance.LifecycleObservedAt),
 		ConfigVersion:        instance.ConfigVersion,
 		ConfigKey:            instance.ConfigKey,
 		ConfigChecksum:       instance.ConfigChecksum,
@@ -216,7 +216,10 @@ type ServerInstance struct {
 	// AdminUserIDs identifies server-scoped administrator accounts.
 	AdminUserIDs []string `json:"adminUserIds" db:"admin_user_ids"`
 	// State is the server lifecycle state.
-	State domain.ServerInstanceState `json:"state" db:"state"`
+	State                   domain.ServerInstanceState `json:"state" db:"state"`
+	LifecycleProcessID      string                     `json:"lifecycleProcessId,omitempty" db:"lifecycle_process_id"`
+	LifecycleObservationSeq uint64                     `json:"lifecycleObservationSeq,omitempty" db:"lifecycle_observation_seq"`
+	LifecycleObservedAt     time.Time                  `json:"lifecycleObservedAt,omitempty" db:"lifecycle_observed_at"`
 	// ConfigVersion is the platform-managed optimistic concurrency version.
 	ConfigVersion int `json:"configVersion" db:"config_version"`
 	// ConfigKey is the logical configuration target, never a host path.
@@ -739,43 +742,49 @@ func remoteAccessFromDomain(remote domain.GamePluginRemoteAccess) GamePluginRemo

 func ServerInstanceFromDomain(instance domain.ServerInstance) ServerInstance {
 	return ServerInstance{
-		ID:                 instance.ID,
-		PluginID:           instance.PluginID,
-		PluginVersion:      instance.PluginVersion,
-		DeploymentTargetID: instance.DeploymentTargetID,
-		RunEndpointID:      instance.RunEndpointID,
-		Name:               instance.Name,
-		OwnerUserID:        instance.OwnerUserID,
-		AdminUserIDs:       domain.CopyStringSlice(instance.AdminUserIDs),
-		State:              instance.State,
-		ConfigVersion:      instance.ConfigVersion,
-		ConfigKey:          instance.ConfigKey,
-		ConfigContent:      instance.ConfigContent,
-		ConfigChecksum:     instance.ConfigChecksum,
-		ConfigUpdatedAt:    instance.ConfigUpdatedAt,
-		CreatedAt:          instance.CreatedAt,
-		UpdatedAt:          instance.UpdatedAt,
+		ID:                      instance.ID,
+		PluginID:                instance.PluginID,
+		PluginVersion:           instance.PluginVersion,
+		DeploymentTargetID:      instance.DeploymentTargetID,
+		RunEndpointID:           instance.RunEndpointID,
+		Name:                    instance.Name,
+		OwnerUserID:             instance.OwnerUserID,
+		AdminUserIDs:            domain.CopyStringSlice(instance.AdminUserIDs),
+		State:                   instance.State,
+		LifecycleProcessID:      instance.LifecycleProcessID,
+		LifecycleObservationSeq: instance.LifecycleObservationSeq,
+		LifecycleObservedAt:     instance.LifecycleObservedAt,
+		ConfigVersion:           instance.ConfigVersion,
+		ConfigKey:               instance.ConfigKey,
+		ConfigContent:           instance.ConfigContent,
+		ConfigChecksum:          instance.ConfigChecksum,
+		ConfigUpdatedAt:         instance.ConfigUpdatedAt,
+		CreatedAt:               instance.CreatedAt,
+		UpdatedAt:               instance.UpdatedAt,
 	}
 }

 func (instance ServerInstance) ToDomain() domain.ServerInstance {
 	return domain.ServerInstance{
-		ID:                 instance.ID,
-		PluginID:           instance.PluginID,
-		PluginVersion:      instance.PluginVersion,
-		DeploymentTargetID: instance.DeploymentTargetID,
-		RunEndpointID:      instance.RunEndpointID,
-		Name:               instance.Name,
-		OwnerUserID:        instance.OwnerUserID,
-		AdminUserIDs:       domain.CopyStringSlice(instance.AdminUserIDs),
-		State:              instance.State,
-		ConfigVersion:      instance.ConfigVersion,
-		ConfigKey:          instance.ConfigKey,
-		ConfigContent:      instance.ConfigContent,
-		ConfigChecksum:     instance.ConfigChecksum,
-		ConfigUpdatedAt:    instance.ConfigUpdatedAt,
-		CreatedAt:          instance.CreatedAt,
-		UpdatedAt:          instance.UpdatedAt,
+		ID:                      instance.ID,
+		PluginID:                instance.PluginID,
+		PluginVersion:           instance.PluginVersion,
+		DeploymentTargetID:      instance.DeploymentTargetID,
+		RunEndpointID:           instance.RunEndpointID,
+		Name:                    instance.Name,
+		OwnerUserID:             instance.OwnerUserID,
+		AdminUserIDs:            domain.CopyStringSlice(instance.AdminUserIDs),
+		State:                   instance.State,
+		LifecycleProcessID:      instance.LifecycleProcessID,
+		LifecycleObservationSeq: instance.LifecycleObservationSeq,
+		LifecycleObservedAt:     instance.LifecycleObservedAt,
+		ConfigVersion:           instance.ConfigVersion,
+		ConfigKey:               instance.ConfigKey,
+		ConfigContent:           instance.ConfigContent,
+		ConfigChecksum:          instance.ConfigChecksum,
+		ConfigUpdatedAt:         instance.ConfigUpdatedAt,
+		CreatedAt:               instance.CreatedAt,
+		UpdatedAt:               instance.UpdatedAt,
 	}
 }

@@ -63,7 +63,7 @@ Platform-owned Run distribution builds embed an autonomous lifecycle plan for th

 The plan is build input for the generated package, not a machine-side job-channel payload. Generated Run registration must not be treated as a trigger to enqueue `process.start`, `process.install`, or `process.status` work; Platform state converges from Run heartbeats, logs, lifecycle reports, supervised process facts, and terminal job/report messages. Platform and Run must not add game-specific hardcoding to interpret the plan.

-Autonomous lifecycle reports use `POST /api/v1/run/lifecycle/report` with the active Run session and signed envelope when required. The route accepts only bounded terminal lifecycle facts for `process.install`, `process.start`, `process.stop`, or `process.status`; it validates the server/run binding, records audit evidence, and projects server state from Run-reported process facts without creating or completing a Platform job.
+Autonomous lifecycle reports use `POST /api/v1/run/lifecycle/report` with the active Run session and signed envelope when required. The route accepts only bounded terminal lifecycle facts for `process.install`, `process.start`, `process.stop`, or `process.status`; it validates the server/run binding, records audit evidence, and projects server state from Run-reported process facts without creating or completing a Platform job. A managed-process report includes an opaque `managedProcessId`, monotonic `observationSeq`, and `observedAt`; retries are idempotent and a lower sequence cannot regress a newer fact for that process.

 ## Log Ingest

@@ -77,6 +77,7 @@ Named log DTOs:

 - `LogBatchIngestRequest`
 - `LogBatchIngestResponse`
+- `RunLogStreamProgressRequest` / `RunLogStreamProgressResponse`: signed Run-only sequence recovery for a server-bound `run.<endpoint>.<server>.*` stream. The response contains only the latest acknowledged sequence.
 - `LogEntry`
 - `LogStreamCursorRequest`
 - `LogStreamCursorResponse`
@@ -4,6 +4,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"

 	"browser.local/platform/domain"
 	"browser.local/platform/repo"
@@ -464,6 +465,30 @@ func TestCoreServiceRunLifecycleReportProjectsGeneratedRunFacts(t *testing.T) {
 	}
 }

+func TestCoreServiceRejectsStaleManagedProcessObservation(t *testing.T) {
+	svc := newTestCoreService()
+	plugin := createGeneratedRunStatusPlugin(t, svc)
+	instance := domain.ServerInstance{ID: "managed-observation-order", PluginID: plugin.ID, PluginVersion: plugin.Version, RunEndpointID: dedicatedRunEndpointID("managed-observation-order"), Name: "Managed Observation Order", State: domain.ServerInstanceStateDraft, ConfigVersion: 1}
+	if err := svc.store.ServerInstances().Create(instance); err != nil {
+		t.Fatalf("create server: %v", err)
+	}
+	registered := registerGeneratedRunForStatusTest(t, svc, instance, plugin.ID)
+	observedAt := time.Date(2026, 8, 7, 10, 0, 0, 0, time.UTC)
+	report := func(sequence uint64, state string, classification string) {
+		t.Helper()
+		if _, err := svc.ReportRunLifecycle(domain.RunLifecycleReport{RunEndpointID: instance.RunEndpointID, SessionToken: registered.SessionToken, ServerInstanceID: instance.ID, Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100}, ManagedProcessID: "sha256:managed-process", ObservationSeq: sequence, ObservedAt: observedAt.Add(time.Duration(sequence) * time.Second), ExecutionResult: domain.JobExecutionResult{Kind: "process", ProcessState: state, ExitClassification: classification}}); err != nil {
+			t.Fatalf("report sequence %d: %v", sequence, err)
+		}
+	}
+	report(1, "running", "")
+	report(2, "exited", "unexpected-exit")
+	report(1, "running", "")
+	stored, err := svc.GetServerInstance(instance.ID)
+	if err != nil || stored.State != domain.ServerInstanceStateFailed || stored.LifecycleObservationSeq != 2 {
+		t.Fatalf("stale running observation must not regress exit projection: server=%+v err=%v", stored, err)
+	}
+}
+
 func TestCoreServiceGeneratedRunRegistrationDoesNotDispatchStatusReconciliation(t *testing.T) {
 	svc := newTestCoreService()
 	plugin := createGeneratedRunStatusPlugin(t, svc)
@@ -219,6 +219,36 @@ func (svc *CoreService) QueryLogStream(query domain.LogStreamCursorQuery) (domai
 	}), nil
 }

+func (svc *CoreService) GetRunLogStreamProgress(request domain.RunLogStreamProgress) (domain.RunLogStreamProgressResult, error) {
+	if err := validator.ValidateRunLogStreamProgress(request); err != nil {
+		return domain.RunLogStreamProgressResult{}, err
+	}
+	if _, err := svc.validatedRunSession(request.RunEndpointID, request.SessionToken); err != nil {
+		return domain.RunLogStreamProgressResult{}, err
+	}
+	instance, err := svc.store.ServerInstances().Get(request.ServerInstanceID)
+	if err != nil {
+		return domain.RunLogStreamProgressResult{}, err
+	}
+	if instance.RunEndpointID != request.RunEndpointID {
+		return domain.RunLogStreamProgressResult{}, validationError("runEndpointId must match server instance")
+	}
+	if !strings.HasPrefix(request.LogStreamID, "run."+request.RunEndpointID+"."+request.ServerInstanceID+".") {
+		return domain.RunLogStreamProgressResult{}, validationError("logStreamId is not a bound run stream")
+	}
+	latest := uint64(0)
+	stream, err := svc.store.LogStreams().Get(request.LogStreamID)
+	if err == nil {
+		if stream.ServerInstanceID != instance.ID {
+			return domain.RunLogStreamProgressResult{}, validationError("logStreamId does not belong to server instance")
+		}
+		latest = stream.LatestSeq
+	} else if !errors.Is(err, repo.ErrNotFound) {
+		return domain.RunLogStreamProgressResult{}, err
+	}
+	return domain.RunLogStreamProgressResult{Accepted: true, RunEndpointID: request.RunEndpointID, ServerInstanceID: instance.ID, LogStreamID: request.LogStreamID, LatestSeq: latest, ServerTime: svc.now()}, nil
+}
+
 func validateLogBatchStream(batch domain.LogBatchIngest, stream domain.LogStream) error {
 	if stream.ID != batch.LogStreamID {
 		return validationError("logStreamId must match stream")
@@ -39,6 +39,26 @@ func TestCoreServiceIngestsLogBatchAndQueriesCursor(t *testing.T) {
 	}
 }

+func TestCoreServiceReturnsBoundRunLogStreamProgress(t *testing.T) {
+	svc, sessionToken := newRegisteredLogIngestService(t)
+	streamID := "run.run-local.server-1.stdout"
+	if _, err := svc.CreateLogStream(domain.LogStream{ID: streamID, ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stdout", StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"}); err != nil {
+		t.Fatalf("create run stream: %v", err)
+	}
+	batch := validLogBatch(t, sessionToken, 1, 2)
+	batch.LogStreamID = streamID
+	if _, err := svc.IngestLogBatch(batch); err != nil {
+		t.Fatalf("ingest run stream: %v", err)
+	}
+	progress, err := svc.GetRunLogStreamProgress(domain.RunLogStreamProgress{RunEndpointID: "run-local", SessionToken: sessionToken, ServerInstanceID: "server-1", LogStreamID: streamID})
+	if err != nil || !progress.Accepted || progress.LatestSeq != 2 {
+		t.Fatalf("unexpected progress: %+v err=%v", progress, err)
+	}
+	if _, err := svc.GetRunLogStreamProgress(domain.RunLogStreamProgress{RunEndpointID: "run-local", SessionToken: sessionToken, ServerInstanceID: "server-1", LogStreamID: "run.run-local.other.stdout"}); err == nil {
+		t.Fatal("expected progress scope validation failure")
+	}
+}
+
 func TestCoreServicePublishesLogEventsForAcceptedBatch(t *testing.T) {
 	svc, sessionToken := newRegisteredLogIngestService(t)
 	createLogStreamFixture(t, svc)
@@ -209,6 +209,7 @@ type Core interface {
 	SubscribeLogEvents(string) (LogEventSubscription, error)
 	SubscribeLogEventsForSession(string, string) (LogEventSubscription, error)
 	IngestLogBatch(domain.LogBatchIngest) (domain.LogBatchIngestResult, error)
+	GetRunLogStreamProgress(domain.RunLogStreamProgress) (domain.RunLogStreamProgressResult, error)
 	QueryLogStream(domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error)
 	ListGamePlayersForSession(string, domain.GamePlayerFilter) ([]domain.GamePlayer, error)
 	GetGamePlayerProfileForSession(string, string) (domain.GamePlayerProfile, error)
@@ -29,8 +29,20 @@ func (svc *CoreService) ReportRunLifecycle(report domain.RunLifecycleReport) (do

 	stamp := svc.now()
 	nextState, projected := lifecycleProjectedState(report.Capability, report.State, report.ExecutionResult)
+	if lifecycleObservationIsStale(instance, report) {
+		projected = false
+		nextState = instance.State
+	}
 	if projected {
 		instance.State = nextState
+		if report.ManagedProcessID != "" {
+			instance.LifecycleProcessID = report.ManagedProcessID
+			instance.LifecycleObservationSeq = report.ObservationSeq
+			instance.LifecycleObservedAt = report.ObservedAt
+			if instance.LifecycleObservedAt.IsZero() {
+				instance.LifecycleObservedAt = stamp
+			}
+		}
 		instance.UpdatedAt = stamp
 		if err := validator.ValidateServerInstance(instance); err != nil {
 			return domain.RunLifecycleReportResult{}, err
@@ -49,6 +61,16 @@ func (svc *CoreService) ReportRunLifecycle(report domain.RunLifecycleReport) (do
 	return domain.CopyRunLifecycleReportResult(domain.RunLifecycleReportResult{Accepted: true, RunEndpointID: report.RunEndpointID, ServerInstanceID: report.ServerInstanceID, ProjectedState: nextState, ServerTime: stamp}), nil
 }

+func lifecycleObservationIsStale(instance domain.ServerInstance, report domain.RunLifecycleReport) bool {
+	if report.ManagedProcessID == "" || instance.LifecycleProcessID == "" {
+		return false
+	}
+	if report.ManagedProcessID == instance.LifecycleProcessID {
+		return report.ObservationSeq <= instance.LifecycleObservationSeq
+	}
+	return !report.ObservedAt.IsZero() && !instance.LifecycleObservedAt.IsZero() && report.ObservedAt.Before(instance.LifecycleObservedAt)
+}
+
 func lifecycleReportSummary(report domain.RunLifecycleReport, projectedState domain.ServerInstanceState, projected bool) string {
 	for _, candidate := range []string{report.ExecutionResult.AuditSummary, report.Progress.Message, report.Message, report.ErrorCode} {
 		if strings.TrimSpace(candidate) != "" {
@@ -69,6 +69,12 @@ func ValidateRunLifecycleReport(report domain.RunLifecycleReport) error {
 	violations = appendProgressViolations(violations, report.Progress)
 	violations = appendMessageLength(violations, "message", report.Message)
 	violations = appendMessageLength(violations, "errorCode", report.ErrorCode)
+	if report.ManagedProcessID != "" && report.ObservationSeq == 0 {
+		violations = append(violations, "observationSeq is required with managedProcessId")
+	}
+	if report.ObservationSeq > 0 && report.ManagedProcessID == "" {
+		violations = append(violations, "managedProcessId is required with observationSeq")
+	}
 	if len([]byte(report.ExecutionResult.Content)) > maxJobChannelMessageLength*256 {
 		violations = append(violations, "executionResult.content is too large")
 	}
@@ -87,6 +87,15 @@ func ValidateLogStreamCursorQuery(query domain.LogStreamCursorQuery) error {
 	return finish(violations)
 }

+func ValidateRunLogStreamProgress(request domain.RunLogStreamProgress) error {
+	var violations []string
+	violations = appendRequired(violations, "runEndpointId", request.RunEndpointID)
+	violations = appendRequired(violations, "sessionToken", request.SessionToken)
+	violations = appendRequired(violations, "serverInstanceId", request.ServerInstanceID)
+	violations = appendRequired(violations, "logStreamId", request.LogStreamID)
+	return finish(violations)
+}
+
 func LogEntriesChecksum(entries []domain.LogEntry) (string, error) {
 	stable := make([]logEntryChecksumBody, len(entries))
 	for i, entry := range entries {