fix(taskruntime): prevent duplicate stream replay

backend/internal/biz/conversation/impl/chat.go

@@ -30,6 +30,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -820,6 +821,7 @@ func (s *Service) buildChatRequest(
 		Model:             resolveAgentModel(singleAgent),
 		AgentRole:         agentRole,
 		NeedUpdateTitle:   conversation.Title == DefaultConversationTitle,
+		SubmissionId:      uuid.NewString(),
 	}
 }
 

backend/internal/biz/conversation/impl/chat_test.go

@@ -27,6 +27,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"sico-backend/internal/biz/agent"
 	singleagententity "sico-backend/internal/entity/agent/singleagent"
@@ -277,3 +279,74 @@ func TestReconnect(t *testing.T) {
 		require.Equal(t, "done", sentEvents[0].Event)
 	})
 }
+
+func TestShouldRetryCoreStreamChatStopsAfterOutput(t *testing.T) {
+	connection := &ChatConnection{}
+	err := status.Error(codes.Unavailable, "core disconnected")
+
+	require.True(t, shouldRetryCoreStreamChat(err, 1, connection))
+	connection.sentSeq = 1
+	require.False(t, shouldRetryCoreStreamChat(err, 1, connection))
+}
+
+// unavailableThenOKChatClient fails the first failAttempts StreamChat calls with
+// codes.Unavailable and records the submission id carried by every attempt.
+type unavailableThenOKChatClient struct {
+	conversationrpc.ChatServiceClient
+	failAttempts  int
+	calls         int
+	submissionIDs []string
+}
+
+func (m *unavailableThenOKChatClient) StreamChat(
+	ctx context.Context,
+	in *conversationdto.ChatRequest,
+	opts ...grpc.CallOption,
+) (*conversationdto.ChatDirectResponse, error) {
+	m.calls++
+	m.submissionIDs = append(m.submissionIDs, in.GetSubmissionId())
+	if m.calls <= m.failAttempts {
+		return nil, status.Error(codes.Unavailable, "core is draining")
+	}
+	return &conversationdto.ChatDirectResponse{}, nil
+}
+
+func newTestChatRequest(service *Service) *conversationdto.ChatRequest {
+	return service.buildChatRequest(
+		context.Background(),
+		&conversationdto.ChatRequestHttp{AgentInstanceID: 1, Message: "hi"},
+		"alice",
+		&singleagentpb.SingleAgent{},
+		nil,
+		&conventity.Conversation{ID: 7},
+		3,
+		nil,
+		nil,
+	)
+}
+
+func TestBuildChatRequestAssignsUniqueSubmissionID(t *testing.T) {
+	service := newTestConversationService()
+
+	first := newTestChatRequest(service)
+	second := newTestChatRequest(service)
+
+	require.NotEmpty(t, first.GetSubmissionId())
+	require.NotEqual(t, first.GetSubmissionId(), second.GetSubmissionId())
+}
+
+func TestStreamChatRetryReusesSubmissionID(t *testing.T) {
+	chatClient := &unavailableThenOKChatClient{failAttempts: 1}
+	service := newTestConversationService()
+	service.chatClient = chatClient
+	chatReq := newTestChatRequest(service)
+
+	err := service.streamChatWithUnavailableRetry(context.Background(), &ChatConnection{}, chatReq)
+
+	require.NoError(t, err)
+	require.Equal(t, 2, chatClient.calls)
+	require.NotEmpty(t, chatClient.submissionIDs[0])
+	// A transport-level retry must keep the identity so core treats the second
+	// attempt as a replay of the same submission rather than a new one.
+	require.Equal(t, chatClient.submissionIDs[0], chatClient.submissionIDs[1])
+}

backend/internal/store/taskruntime/internal/dal/errors.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 
 	"github.com/go-sql-driver/mysql"
+	"gorm.io/gorm"
 )
 
 // Stable error message tokens. Python clients translate by gRPC status code
@@ -49,6 +50,9 @@ func isDuplicateKey(err error) bool {
 	if err == nil {
 		return false
 	}
+	if errors.Is(err, gorm.ErrDuplicatedKey) {
+		return true
+	}
 
 	var mysqlErr *mysql.MySQLError
 	if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlDuplicateErrNum {

backend/internal/store/taskruntime/internal/dal/errors_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/go-sql-driver/mysql"
+	"gorm.io/gorm"
 )
 
 func TestIsStaleTokenIdentifiesWrappedErrors(t *testing.T) {
@@ -47,6 +48,15 @@ func TestIsDuplicateKeyFallsBackOnMessage(t *testing.T) {
 	}
 }
 
+func TestIsDuplicateKeyHandlesTranslatedGORMError(t *testing.T) {
+	if !isDuplicateKey(gorm.ErrDuplicatedKey) {
+		t.Fatal("gorm.ErrDuplicatedKey should be a duplicate key")
+	}
+	if !isDuplicateKey(fmt.Errorf("create run: %w", gorm.ErrDuplicatedKey)) {
+		t.Fatal("wrapped gorm.ErrDuplicatedKey should be a duplicate key")
+	}
+}
+
 func TestIsRetryableTransactionErrorHandlesMySQLDeadlocks(t *testing.T) {
 	if !isRetryableTransactionError(&mysql.MySQLError{Number: mysqlDeadlockErrNum}) {
 		t.Fatal("MySQLError 1213 should be retryable")

core/app/biz/chat/chat.py

@@ -204,6 +204,22 @@ async def run_stream(
                         len(prior_partial_messages),
                         exc_info=True,
                     )
+                    if self._tool_context is not None and self._tool_context.task_runtime_batch_ids:
+                        _LOGGER.warning(
+                            "chat_client_stream_retry_suppressed_after_task_submission "
+                            "agent_instance_id=%s conversation_id=%s turn_id=%s batch_ids=%s",
+                            self._agent_instance_id,
+                            conversation_id,
+                            options.turn_id,
+                            self._tool_context.task_runtime_batch_ids,
+                        )
+                        await queue.put(
+                            build_error_response(
+                                "Chat response generation stopped, but the submitted tasks are still running. "
+                                "Their progress and results remain available in the current plan."
+                            )
+                        )
+                        raise
                     error_text = f"Error with chat agent attempt {attempt}/{options.max_attempts}: {str(exc)}"
                     error_response = build_error_response(error_text)
                     error_message = Message(role="assistant", contents=[Content.from_text(error_text)])

core/app/biz/chat/service.py

@@ -286,6 +286,7 @@ async def on_plan_update(plan: Plan):
             response_queue=response_queue,
             plan_editor=plan_editor,
             raw_user_message=chat_request.message.content,
+            submission_id=chat_request.submission_id,
         )
 
         sequence_id = 0

core/app/biz/task_runtime/config.py

@@ -30,19 +30,19 @@
 * :func:`_resolve_positive_env` rejects non-positive values back to the default
   (concurrency ceilings).
 
-=======================================  ========  =====  ========
-Environment variable                     Default   Floor  Strategy
-=======================================  ========  =====  ========
-TASK_RUNTIME_REUSE_WAIT_TIMEOUT_SECONDS  policy+30  1     clamp
-TASK_RUNTIME_SANDBOX_RELEASE_ATTEMPTS    3         1      clamp
-TASK_RUNTIME_STALE_RUN_AFTER_MS          180000    0      clamp
-TASK_RUNTIME_RECONCILE_INTERVAL_SECONDS  30        1      clamp
-TASK_RUNTIME_HEARTBEAT_INTERVAL_SECONDS  30        1      clamp
-TASK_RUNTIME_REVERSE_RPC_TIMEOUT_SECONDS 20        1      clamp
-TASK_RUNTIME_MAX_CONCURRENCY             scheduler 1      reject
-TASK_RUNTIME_K8S_POD_CONCURRENCY         10        1      reject
-TASK_RUNTIME_DOCKER_CONCURRENCY          10        1      reject
-=======================================  ========  =====  ========
+===================================================  ==========  ======  ========
+Environment variable                                 Default     Floor   Strategy
+===================================================  ==========  ======  ========
+TASK_RUNTIME_REUSE_WAIT_TIMEOUT_SECONDS              policy+30   1       clamp
+TASK_RUNTIME_REPLAY_MATERIALIZATION_TIMEOUT_SECONDS  30          1       clamp
+TASK_RUNTIME_SANDBOX_RELEASE_ATTEMPTS                3           1       clamp
+TASK_RUNTIME_STALE_RUN_AFTER_MS                      180000      0       clamp
+TASK_RUNTIME_HEARTBEAT_INTERVAL_SECONDS              30          1       clamp
+TASK_RUNTIME_REVERSE_RPC_TIMEOUT_SECONDS             20          1       clamp
+TASK_RUNTIME_MAX_CONCURRENCY                         scheduler   1       reject
+TASK_RUNTIME_K8S_POD_CONCURRENCY                     10          1       reject
+TASK_RUNTIME_DOCKER_CONCURRENCY                      10          1       reject
+===================================================  ==========  ======  ========
 
 ``clamp`` raises sub-floor values up to ``floor``; ``reject`` reverts any value
 below 1 back to the default.
@@ -93,8 +93,8 @@ def _stale_run_after_ms() -> int:
     return _resolve_clamped_env("TASK_RUNTIME_STALE_RUN_AFTER_MS", 180000, floor=0)
 
 
-def _task_runtime_reconcile_interval_seconds() -> int:
-    return _resolve_clamped_env("TASK_RUNTIME_RECONCILE_INTERVAL_SECONDS", 30, floor=1)
+def _replay_run_materialization_timeout_seconds() -> int:
+    return _resolve_clamped_env("TASK_RUNTIME_REPLAY_MATERIALIZATION_TIMEOUT_SECONDS", 30, floor=1)
 
 
 def _task_runtime_heartbeat_interval_seconds() -> int:

core/app/biz/task_runtime/context.py

@@ -18,7 +18,7 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-"""Per-turn identity + plan-editor envelope passed into TaskManager.
+"""Per-submission identity + plan-editor envelope passed into TaskManager.
 
 Used by ``TaskManager.submit_prepared`` as the narrow per-turn context for a
 single chat turn. Trusted handoff — fields are validated upstream by the
@@ -29,6 +29,7 @@
 * identity fields (``username`` / ``agent_*`` / ``project_id`` /
   ``conversation_id`` / ``turn_id``)
 * ``plan_editor`` — streaming channel for lifecycle / tool-call UI updates
+* ``submission_id`` — stable identity for one logical task submission
 * ``task_runtime_batch_ids`` — mutable list shared with the source
   ``ToolContext`` tracking the batches this turn created.
 
@@ -64,22 +65,35 @@ class TurnContext:
     """Live channel for streaming task lifecycle updates to the frontend plan
     UI. Manager writes; tools/skills do not own this reference."""
 
+    submission_id: str = ""
+    """Stable identity for one logical task submission across transport retries."""
+
+    submission_source: str = ""
+    """Stable producer identity included in replay fingerprint validation."""
+
     task_runtime_batch_ids: list[str] = field(default_factory=list)
     """Append-only list of batch ids submitted during this turn. Mutated by
     the orchestrator on every successful ``submit_prepared``. When constructed via
     :meth:`from_tool_context` the list reference is shared with the source
     ``ToolContext`` so mutations propagate."""
 
     @classmethod
-    def from_tool_context(cls, tc: ToolContext) -> TurnContext:
+    def from_tool_context(
+        cls,
+        tc: ToolContext,
+        *,
+        submission_id: str = "",
+        submission_source: str = "",
+    ) -> TurnContext:
         """Narrow the wide ``ToolContext`` used inside chat tools into the
         per-turn context required by ``TaskManager.submit_prepared``.
 
         Drops ``response_queue`` / ``all_tools`` — those belong to tool
         execution, not to task scheduling. ``task_runtime_batch_ids`` shares
         the list reference so mutations from inside the manager propagate
-        back to the caller. Coerces ``agent_instance_id`` ``None`` to ``0``
-        so downstream code does not need a null check on every read."""
+        back to the caller. ``submission_id`` can be narrowed to one delegate
+        invocation. Coerces ``agent_instance_id`` ``None`` to ``0`` so downstream
+        code does not need a null check on every read."""
         return cls(
             username=tc.username,
             agent_id=tc.agent_id,
@@ -88,6 +102,8 @@ def from_tool_context(cls, tc: ToolContext) -> TurnContext:
             conversation_id=tc.conversation_id,
             turn_id=tc.turn_id,
             plan_editor=tc.plan_editor,
+            submission_id=submission_id or tc.submission_id,
+            submission_source=submission_source,
             task_runtime_batch_ids=tc.task_runtime_batch_ids,
         )
 

core/app/biz/task_runtime/models.py

@@ -207,10 +207,9 @@ class TaskSpec(BaseModel):
     # hand-off). Gaps are allowed: distinct values are ordered ascending into
     # execution waves.
     stage: int = 0
-    # Optional caller-supplied idempotency key. When set, retrying with the
-    # exact same key (within the same turn) returns the prior run instead of
-    # creating a new one. When empty, the runtime derives a stable key from
-    # the task contents.
+    # Optional caller-supplied task identity within one logical submission.
+    # The runtime scopes it with the submission id and batch position, so a new
+    # user-requested submission still executes while transport replay reuses.
     idempotency_key: str = ""
 
     @field_validator("task_id", "title")
@@ -451,22 +450,23 @@ def from_result(cls, result: BatchResult, *, max_success_items: int = 3) -> "Bat
         )
 
 
-def compute_idempotency_key(conversation_id: int, turn_id: int, batch_item_index: int, task: TaskSpec) -> str:
-    """Derive a stable idempotency key for a task within a conversation turn.
+def compute_idempotency_key(submission_id: str, batch_item_index: int, task: TaskSpec) -> str:
+    """Derive a stable task key within one logical submission.
 
-    The key intentionally excludes fields that change between retries (run/tool-call IDs,
-    timestamps, attempt counts). If the caller supplied an explicit ``task.idempotency_key``,
-    we trust it verbatim so callers can dedupe across turns. Otherwise we hash the
-    canonical contents of the task plus its position within the batch.
+    Transport retries preserve ``submission_id`` and therefore reuse existing
+    work. An intentional rerun receives a new submission id and executes again,
+    even when the task contents are unchanged.
     """
+    submission_id = submission_id.strip()
+    if not submission_id:
+        raise ValueError("submission_id is required")
+
     explicit = task.idempotency_key.strip()
-    if explicit:
-        return explicit
     payload = {
-        "conversation_id": conversation_id,
-        "turn_id": turn_id,
+        "schema_version": 2,
+        "submission_id": submission_id,
         "batch_item_index": batch_item_index,
-        "task": task.model_dump(
+        "task": {"explicit_idempotency_key": explicit} if explicit else task.model_dump(
             mode="json",
             exclude={"task_id", "idempotency_key", "metadata"},
             exclude_none=True,

core/app/biz/task_runtime/run_coordinator.py

@@ -187,6 +187,12 @@ async def _mark_run_cancelled(self, ctx: TurnContext, run: TaskRun, reason: str)
     # -- reuse path ---------------------------------------------------------
 
     async def _wait_for_existing_run(self, ctx: TurnContext, run: TaskRun) -> TaskResult:
+        # This path is an observer, not a replacement owner: claiming a queued
+        # run can allocate duplicate sandbox resources before fencing resolves,
+        # and a running external process cannot be resumed safely. If the owner
+        # died, its batch heartbeat expires and the stale reconciler persists a
+        # FAILED/BLOCKED result for us to observe. A local timeout is surfaced as
+        # BLOCKED rather than risking duplicate side effects.
         started_at = _now_ms()
         wait_timeout = _reuse_wait_timeout_seconds(run)
         deadline = time.monotonic() + wait_timeout