feat: persist fenced client command events

config/buf-lint-baseline.txt

@@ -1,12 +1,12 @@
-protos/rvbox/v1/common.proto:262:5:Enum value name "INVALID_ARGUMENT" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:265:5:Enum value name "INVALID_ARGUMENT" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:263:5:Enum value name "NOT_FOUND" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:266:5:Enum value name "NOT_FOUND" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:264:5:Enum value name "OFFLINE" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:267:5:Enum value name "OFFLINE" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:265:5:Enum value name "CAPACITY_EXHAUSTED" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:268:5:Enum value name "CAPACITY_EXHAUSTED" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:266:5:Enum value name "CONFLICT" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:269:5:Enum value name "CONFLICT" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:267:5:Enum value name "UNSUPPORTED" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:270:5:Enum value name "UNSUPPORTED" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:268:5:Enum value name "PROTOCOL_ERROR" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:271:5:Enum value name "PROTOCOL_ERROR" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:269:5:Enum value name "TRANSIENT" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:272:5:Enum value name "TRANSIENT" should be prefixed with "CODE_".
-protos/rvbox/v1/common.proto:270:5:Enum value name "INTERNAL" should be prefixed with "CODE_".
+protos/rvbox/v1/common.proto:273:5:Enum value name "INTERNAL" should be prefixed with "CODE_".
 protos/rvbox/v1/common.proto:32:3:Enum value name "SHELL_SH" should be prefixed with "SHELL_TYPE_".
 protos/rvbox/v1/common.proto:33:3:Enum value name "SHELL_BASH" should be prefixed with "SHELL_TYPE_".
 protos/rvbox/v1/common.proto:34:3:Enum value name "SHELL_CMD" should be prefixed with "SHELL_TYPE_".

gen/go/rvbox/v1/common.pb.go

@@ -1830,6 +1830,9 @@ type CommandEvent struct {
 	//	*CommandEvent_OutputTruncation
 	//	*CommandEvent_OutputIncomplete
 	Payload isCommandEvent_Payload `protobuf_oneof:"payload"`
+	// SHA-256 of this event with this field cleared. It makes retransmission
+	// conflicts detectable without trusting a reused event_seq.
+	ImmutableEventSha256 []byte `protobuf:"bytes,12,opt,name=immutable_event_sha256,json=immutableEventSha256,proto3" json:"immutable_event_sha256,omitempty"`
 	unknownFields        protoimpl.UnknownFields
 	sizeCache            protoimpl.SizeCache
 }
@@ -1964,6 +1967,13 @@ func (x *CommandEvent) GetOutputIncomplete() *OutputIncomplete {
 	return nil
 }
 
+func (x *CommandEvent) GetImmutableEventSha256() []byte {
+	if x != nil {
+		return x.ImmutableEventSha256
+	}
+	return nil
+}
+
 type isCommandEvent_Payload interface {
 	isCommandEvent_Payload()
 }
@@ -2211,7 +2221,7 @@ const file_rvbox_v1_common_proto_rawDesc = "" +
 	"\x12ScriptUploadStatus\x12%\n" +
 	"\x0ereceived_bytes\x18\x01 \x01(\x04R\rreceivedBytes\x12\x1a\n" +
 	"\bcomplete\x18\x02 \x01(\bR\bcomplete\x12\x16\n" +
-	"\x06detail\x18\x03 \x01(\tR\x06detail\"\x91\x05\n" +
+	"\x06detail\x18\x03 \x01(\tR\x06detail\"\xc7\x05\n" +
 	"\fCommandEvent\x12\x1d\n" +
 	"\n" +
 	"issue_uuid\x18\x01 \x01(\tR\tissueUuid\x12\x1b\n" +
@@ -2226,7 +2236,8 @@ const file_rvbox_v1_common_proto_rawDesc = "" +
 	"\rscript_status\x18\t \x01(\v2\x1c.rvbox.v1.ScriptUploadStatusH\x00R\fscriptStatus\x12I\n" +
 	"\x11output_truncation\x18\n" +
 	" \x01(\v2\x1a.rvbox.v1.OutputTruncationH\x00R\x10outputTruncation\x12I\n" +
-	"\x11output_incomplete\x18\v \x01(\v2\x1a.rvbox.v1.OutputIncompleteH\x00R\x10outputIncompleteB\t\n" +
+	"\x11output_incomplete\x18\v \x01(\v2\x1a.rvbox.v1.OutputIncompleteH\x00R\x10outputIncomplete\x124\n" +
+	"\x16immutable_event_sha256\x18\f \x01(\fR\x14immutableEventSha256B\t\n" +
 	"\apayload\"\xb2\x03\n" +
 	"\fControlError\x12/\n" +
 	"\x04code\x18\x01 \x01(\x0e2\x1b.rvbox.v1.ControlError.CodeR\x04code\x12\x18\n" +

internal/agentproto/validate.go

@@ -220,12 +220,16 @@ func validateCommandAccepted(accepted *rvboxv1.CommandAccepted, limits Limits) e
 }
 
 func validateCommandEvent(event *rvboxv1.CommandEvent, limits Limits) error {
-	if event == nil || event.Payload == nil || event.ObservedAt == nil || event.ObservedAt.CheckValid() != nil {
+	if event == nil || event.Payload == nil || event.ObservedAt == nil || event.ObservedAt.CheckValid() != nil || len(event.ImmutableEventSha256) != sha256.Size {
 		return fmt.Errorf("%w: invalid command event", ErrInvalidEnvelope)
 	}
 	if err := validateIssueAndSequence(event.IssueUuid, event.EventSeq); err != nil {
 		return err
 	}
+	digest, err := CommandEventDigest(event)
+	if err != nil || string(digest[:]) != string(event.ImmutableEventSha256) {
+		return fmt.Errorf("%w: command event immutable digest mismatch", ErrInvalidEnvelope)
+	}
 	switch payload := event.Payload.(type) {
 	case *rvboxv1.CommandEvent_Output:
 		_, err := DecodeOutputChunk(payload.Output, limits.MaxRawChunkBytes)
@@ -265,6 +269,22 @@ func validateCommandEvent(event *rvboxv1.CommandEvent, limits Limits) error {
 	return nil
 }
 
+// CommandEventDigest is the canonical retransmission digest: protobuf bytes of
+// the event with its digest field cleared. Producers populate the field with
+// this value; consumers verify it before using event_seq for idempotency.
+func CommandEventDigest(event *rvboxv1.CommandEvent) ([sha256.Size]byte, error) {
+	if event == nil {
+		return [sha256.Size]byte{}, ErrInvalidEnvelope
+	}
+	copyEvent := proto.Clone(event).(*rvboxv1.CommandEvent)
+	copyEvent.ImmutableEventSha256 = nil
+	encoded, err := proto.MarshalOptions{Deterministic: true}.Marshal(copyEvent)
+	if err != nil {
+		return [sha256.Size]byte{}, err
+	}
+	return sha256.Sum256(encoded), nil
+}
+
 func validateScriptDescriptor(descriptor *rvboxv1.ScriptDescriptor, limits Limits) error {
 	if descriptor == nil || descriptor.Filename == "" || len(descriptor.Filename) > 255 || descriptor.SizeBytes > limits.MaxScriptBytes || len(descriptor.Sha256) != sha256.Size || !utf8.ValidString(descriptor.Filename) {
 		return fmt.Errorf("%w: invalid script descriptor", ErrInvalidExecutionSpec)

internal/agentproto/validate_test.go

@@ -209,7 +209,12 @@ func TestOutputChunkBoundedDecompression_HP_PROTO_05(t *testing.T) {
 func TestCommandEventPayloadValidation_BH_OUTFLOW_01(t *testing.T) {
 	t.Parallel()
 
-	event := &rvboxv1.CommandEvent{IssueUuid: testIssueUUID, EventSeq: 1, ObservedAt: fixedTimestamp(), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}}
+	event := &rvboxv1.CommandEvent{IssueUuid: testIssueUUID, EventSeq: 1, ObservedAt: fixedTimestamp(), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}, ImmutableEventSha256: make([]byte, sha256.Size)}
+	digest, err := CommandEventDigest(event)
+	if err != nil {
+		t.Fatal(err)
+	}
+	event.ImmutableEventSha256 = digest[:]
 	envelope := &rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_CommandEvent{CommandEvent: event}}
 	if err := ValidateEnvelope(envelope, DefaultLimits(), rvboxv1.Platform_PLATFORM_WINDOWS); err != nil {
 		t.Fatalf("valid event: %v", err)

internal/server/session/agent_server.go

@@ -241,7 +241,70 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w
 				server.close(connection, websocket.StatusPolicyViolation, "invalid command acceptance")
 				return
 			}
+			continue
 		}
+		if event := envelope.GetCommandEvent(); event != nil {
+			appendEvent, eventErr := eventAppendFromWire(event, hello.GetClientId(), server.now())
+			if eventErr != nil {
+				server.close(connection, websocket.StatusPolicyViolation, "invalid command event")
+				return
+			}
+			appended, eventErr := server.Store.AppendCommandEvent(sessionContext, appendEvent)
+			if eventErr != nil {
+				server.close(connection, websocket.StatusPolicyViolation, "command event was not accepted")
+				return
+			}
+			ack, eventErr := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: encodedSessionID, SessionGeneration: registration.Generation, Payload: &rvboxv1.AgentEnvelope_EventAck{EventAck: &rvboxv1.EventAck{IssueUuid: event.GetIssueUuid(), ThroughEventSeq: appended.ThroughEventSeq}}})
+			if eventErr != nil || queue.EnqueueControl(Frame{Kind: FrameControl, Payload: ack}) != nil {
+				server.close(connection, websocket.StatusInternalError, "could not acknowledge command event")
+				return
+			}
+		}
+	}
+}
+
+func eventAppendFromWire(event *rvboxv1.CommandEvent, clientID string, receipt time.Time) (store.EventAppend, error) {
+	issue, err := domain.ParseUUIDv7(event.GetIssueUuid())
+	if err != nil {
+		return store.EventAppend{}, err
+	}
+	payload, err := proto.MarshalOptions{Deterministic: true}.Marshal(event)
+	if err != nil {
+		return store.EventAppend{}, err
+	}
+	var immutable [32]byte
+	copy(immutable[:], event.GetImmutableEventSha256())
+	result := store.EventAppend{IssueUUID: [16]byte(issue), ClientID: clientID, EventSeq: event.GetEventSeq(), ObservedUnixNano: event.GetObservedAt().AsTime().UnixNano(), ReceiptUnixNano: receipt.UnixNano(), EventType: eventType(event), Compression: 1, RawLength: uint64(len(payload)), Payload: payload, ImmutableSHA256: immutable}
+	if output := event.GetOutput(); output != nil {
+		result.Stream = uint16(output.GetStream())
+		result.Output = true
+	}
+	if result.EventType == 0 {
+		return store.EventAppend{}, errors.New("unsupported command event")
+	}
+	return result, nil
+}
+
+func eventType(event *rvboxv1.CommandEvent) uint16 {
+	switch event.Payload.(type) {
+	case *rvboxv1.CommandEvent_Lifecycle:
+		return 4
+	case *rvboxv1.CommandEvent_Output:
+		return 5
+	case *rvboxv1.CommandEvent_Resource:
+		return 6
+	case *rvboxv1.CommandEvent_StdinAck:
+		return 7
+	case *rvboxv1.CommandEvent_SignalResult:
+		return 8
+	case *rvboxv1.CommandEvent_ScriptStatus:
+		return 9
+	case *rvboxv1.CommandEvent_OutputTruncation:
+		return 10
+	case *rvboxv1.CommandEvent_OutputIncomplete:
+		return 11
+	default:
+		return 0
 	}
 }
 

internal/server/session/agent_server_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/coder/websocket"
 	rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
+	"github.com/rvbox/rvbox/internal/agentproto"
 	"github.com/rvbox/rvbox/internal/domain"
 	"github.com/rvbox/rvbox/internal/server/store"
 	"google.golang.org/protobuf/proto"
@@ -108,6 +109,19 @@ func TestAgentServerRejectsHostileWireInputs_BH_SES_04(t *testing.T) {
 	}
 }
 
+func TestWireEventAppendCarriesClientBinding_HP_EVENT_01(t *testing.T) {
+	event := &rvboxv1.CommandEvent{IssueUuid: "019c46f1-1d02-7000-8000-000000000096", EventSeq: 1, ObservedAt: timestamppb.Now(), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}}
+	digest, err := agentproto.CommandEventDigest(event)
+	if err != nil {
+		t.Fatal(err)
+	}
+	event.ImmutableEventSha256 = digest[:]
+	appendEvent, err := eventAppendFromWire(event, "client-a", time.Now())
+	if err != nil || appendEvent.ClientID != "client-a" || appendEvent.EventSeq != 1 || appendEvent.EventType != 4 || appendEvent.ImmutableSHA256 != digest {
+		t.Fatalf("wire event append = %#v, %v", appendEvent, err)
+	}
+}
+
 func newTestAgentServer(t *testing.T) (*httptest.Server, func()) {
 	t.Helper()
 	dataDirectory := filepath.Join(t.TempDir(), "store")

internal/server/store/event.go

@@ -33,6 +33,7 @@ func (function FaultInjectorFunc) Checkpoint(name string) error { return functio
 
 type EventAppend struct {
 	IssueUUID        [16]byte
+	ClientID         string
 	EventSeq         uint64
 	ObservedUnixNano int64
 	ReceiptUnixNano  int64
@@ -80,11 +81,17 @@ func (store *Store) AppendCommandEvent(ctx context.Context, event EventAppend) (
 	}
 	var lastSequence, commandOutputCharged, commandCharged, clientCharged, serverCharged, closeoutRemaining uint64
 	var clientID string
-	err = database.QueryRowContext(ctx, `SELECT commands.last_event_seq, commands.output_charged_bytes,
+	query := `SELECT commands.last_event_seq, commands.output_charged_bytes,
 commands.charged_bytes, commands.closeout_remaining_bytes, commands.client_id, clients.charged_bytes,
 storage_counters.command_charged_bytes
 FROM commands JOIN clients ON clients.client_id = commands.client_id
-JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uuid = ?`, event.IssueUUID[:]).Scan(
+JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uuid = ?`
+	arguments := []any{event.IssueUUID[:]}
+	if event.ClientID != "" {
+		query += ` AND commands.client_id = ?`
+		arguments = append(arguments, event.ClientID)
+	}
+	err = database.QueryRowContext(ctx, query, arguments...).Scan(
 		&lastSequence, &commandOutputCharged, &commandCharged, &closeoutRemaining, &clientID, &clientCharged, &serverCharged)
 	if errors.Is(err, sql.ErrNoRows) {
 		return result, ErrCommandNotFound

protos/rvbox/v1/common.proto

@@ -254,6 +254,9 @@ message CommandEvent {
     OutputTruncation output_truncation = 10;
     OutputIncomplete output_incomplete = 11;
   }
+  // SHA-256 of this event with this field cleared. It makes retransmission
+  // conflicts detectable without trusting a reused event_seq.
+  bytes immutable_event_sha256 = 12;
 }
 
 message ControlError {

test/coverage.toml

@@ -218,6 +218,12 @@ layer = "unit"
 status = "implemented"
 tests = ["internal/server/store/command_test.go:TestRecordCommandAcceptanceFencesGeneration_HP_DISPATCH_05"]
 
+[[requirements]]
+id = "HP-EVENT-01"
+layer = "unit"
+status = "implemented"
+tests = ["internal/server/session/agent_server_test.go:TestWireEventAppendCarriesClientBinding_HP_EVENT_01"]
+
 [[requirements]]
 id = "HP-SES-05"
 layer = "integration"