MVP-08: deliver the deterministic local multi-user E2E slice-sessionless-AtomGit

M.github/workflows/ci.yml+1-1文件内容审核中，请稍后刷新重试

MMakefile+5-1文件内容审核中，请稍后刷新重试

MREADME.md+9-2文件内容审核中，请稍后刷新重试

Mcmd/reconciler/main.go+2-1文件内容审核中，请稍后刷新重试

Mcmd/telegram-sender/main.go+16-1文件内容审核中，请稍后刷新重试

Mcmd/worker-runtime/main.go+7-5文件内容审核中，请稍后刷新重试

	@@ -134,6 +134,8 @@ services:
134	TELEGRAM_API_BASE_URL: http://telegram-fake:8081	134	TELEGRAM_API_BASE_URL: http://telegram-fake:8081
135	TELEGRAM_BOT_TOKEN: ${TELEGRAM_FAKE_TOKEN:-local-test-token}	135	TELEGRAM_BOT_TOKEN: ${TELEGRAM_FAKE_TOKEN:-local-test-token}
136	TELEGRAM_SENDER_POLL_INTERVAL: 250ms	136	TELEGRAM_SENDER_POLL_INTERVAL: 250ms
		137	+ TELEGRAM_SENDER_BASE_BACKOFF: ${TELEGRAM_SENDER_BASE_BACKOFF:-250ms}
		138	+ TELEGRAM_SENDER_MAX_BACKOFF: ${TELEGRAM_SENDER_MAX_BACKOFF:-2s}
137	init: true	139	init: true
138	read_only: true	140	read_only: true
139	security_opt:	141	security_opt:
	@@ -209,6 +211,7 @@ services:
209	WORKER_MAX_BLOB_BYTES: ${WORKER_MAX_BLOB_BYTES:-67108864}	211	WORKER_MAX_BLOB_BYTES: ${WORKER_MAX_BLOB_BYTES:-67108864}
210	DETERMINISTIC_HARNESS_TURNS: ${DETERMINISTIC_HARNESS_TURNS:-2}	212	DETERMINISTIC_HARNESS_TURNS: ${DETERMINISTIC_HARNESS_TURNS:-2}
211	DETERMINISTIC_HARNESS_ARTIFACTS: ${DETERMINISTIC_HARNESS_ARTIFACTS:-1}	213	DETERMINISTIC_HARNESS_ARTIFACTS: ${DETERMINISTIC_HARNESS_ARTIFACTS:-1}
		214	+ DETERMINISTIC_HARNESS_FAIL_BEFORE_FIRST_TURN: ${DETERMINISTIC_HARNESS_FAIL_BEFORE_FIRST_TURN:-false}
212	DETERMINISTIC_HARNESS_FAIL_AT_TURN: ${DETERMINISTIC_HARNESS_FAIL_AT_TURN:-0}	215	DETERMINISTIC_HARNESS_FAIL_AT_TURN: ${DETERMINISTIC_HARNESS_FAIL_AT_TURN:-0}
213	DETERMINISTIC_HARNESS_RETRYABLE_FAIL: ${DETERMINISTIC_HARNESS_RETRYABLE_FAIL:-false}	216	DETERMINISTIC_HARNESS_RETRYABLE_FAIL: ${DETERMINISTIC_HARNESS_RETRYABLE_FAIL:-false}
214	init: true	217	init: true

Mdocs/development.md+4-0文件内容审核中，请稍后刷新重试

Mdocs/local-development-stand.md+13-0

	@@ -66,6 +66,7 @@ From a fresh checkout:
66	```sh	66	```sh
67	make dev-up	67	make dev-up
68	make local-integration	68	make local-integration
		69	+make e2e-local
69	make dev-down	70	make dev-down
70	```	71	```
71		72
	@@ -131,6 +132,18 @@ expiry. Worker unit/YDB integration coverage proves tenant-safe materialization,
131	checkpoint resume, lease renewal/loss fencing, cancellation, runtime and turn	132	checkpoint resume, lease renewal/loss fencing, cancellation, runtime and turn
132	limits, exactly-once terminal delivery, and lease-index cleanup.	133	limits, exactly-once terminal delivery, and lease-index cleanup.
133		134
		135	+Run the composed deterministic product slice with:
		136	+
		137	+```sh
		138	+make e2e-local
		139	+```
		140	+
		141	+Unlike the adapter-contract suite, this target drives two Telegram tenants
		142	+through the live control API, reconciler, queue, one-shot worker containers,
		143	+object storage, durable delivery and Telegram capture. It also injects bounded
		144	+queue, worker and Telegram failures. The full scenario list, timing boundary
		145	+and operator queries are documented in [local-e2e.md](local-e2e.md).
		146	+
134	## Apple Silicon	147	## Apple Silicon
135		148
136	The pinned YDB Local image is run as `linux/amd64`, matching YDB's documented	149	The pinned YDB Local image is run as `linux/amd64`, matching YDB's documented

Adocs/local-e2e.md+136-0

@@ -0,0 +1,136 @@
	1	+# Deterministic local end-to-end slice
	2	+
	3	+`make e2e-local` is the black-box proof gate between the individual adapter
	4	+contracts and cloud-dev deployment. It requires Docker Compose but no cloud,
	5	+Telegram, provider, or subscription credential.
	6	+
	7	+## Executed topology
	8	+
	9	+```mermaid
	10	+sequenceDiagram
	11	+ participant Fixture as E2E fixture
	12	+ participant API as control-api
	13	+ participant DB as YDB Local
	14	+ participant Scheduler as reconciler
	15	+ participant Queue as ElasticMQ
	16	+ participant Worker as one-shot worker container
	17	+ participant Blob as MinIO
	18	+ participant Sender as telegram-sender
	19	+ participant Telegram as Telegram fake
	20	+
	21	+ Fixture->>API: signed Telegram update
	22	+ API->>Blob: normalized message and attachments
	23	+ API->>DB: update + run + attempt + manifest + dispatch outbox
	24	+ Scheduler->>DB: tenant-scoped admission + quota reservation + worker job
	25	+ Scheduler->>Queue: tenant_id + run_id
	26	+ Worker->>Queue: receive one dispatch
	27	+ Worker->>DB: fenced lease + checkpoints + usage
	28	+ Worker->>Blob: content-addressed outputs
	29	+ Worker->>DB: terminal state + manifest + delivery outbox
	30	+ Sender->>Telegram: same-chat text and documents
	31	+ Sender->>DB: sent or retry-wait transition
	32	+```
	33	+
	34	+The worker is built and invoked through the Compose `worker` profile. Each
	35	+invocation is read-only, nonroot, concurrency-one, receives one queue message,
	36	+uses a private tmpfs scratch directory, and is removed on exit.
	37	+
	38	+## Command
	39	+
	40	+```sh
	41	+make e2e-local
	42	+```
	43	+
	44	+The command:
	45	+
	46	+1. starts and migrates the pinned local stand;
	47	+2. builds the worker image;
	48	+3. resets only ephemeral Telegram captures and visible queue messages;
	49	+4. runs the tagged black-box suite;
	50	+5. prints service logs automatically when the suite fails.
	51	+
	52	+The stand is intentionally left running for inspection. Use `make dev-down` for
	53	+a non-destructive stop or the guarded `dev-reset` command when local volumes
	54	+must be deleted.
	55	+
	56	+## Automated scenarios
	57	+
	58	+The suite proves:
	59	+
	60	+- two tenants submit interleaved and out-of-order Telegram updates;
	61	+- a duplicate Bot API update maps to one logical run and one dispatch;
	62	+- disconnected subscriptions are blocked, then admitted with provider quota
	63	+ explicitly `unknown` without any API-billing fallback;
	64	+- provider quota `exhausted` and `blocked_until_reset` recover only after an
	65	+ explicit observation/state change;
	66	+- an admitted dispatch whose first queue publication fails is republished by
	67	+ the reconciler after the queue recovers;
	68	+- retryable worker failures before the first checkpoint and after checkpoint
	69	+ one resume without duplicate terminal state;
	70	+- durable cancellation releases the reservation and creates a same-chat reply;
	71	+- a deterministic Telegram 429 produces a durable delivery retry and one
	72	+ logical captured reply;
	73	+- replaying a terminal queue message produces no re-execution, charge, or
	74	+ delivery;
	75	+- output artifact keys and reads remain tenant-scoped;
	76	+- `/new` advances the frontend conversation context epoch explicitly.
	77	+
	78	+The lower-level worker, YDB, S3, queue, ingress, and delivery suites retain
	79	+their focused concurrency, fencing, path traversal, size-limit, and
	80	+cross-tenant negative cases. The E2E suite composes those same production
	81	+adapters rather than replacing them with an in-memory product path.
	82	+
	83	+## Correlation and timing evidence
	84	+
	85	+The test emits one correlation line per input:
	86	+
	87	+```text
	88	+correlation update_id=<id> tenant_id=<opaque> run_id=<opaque> chat_id=<synthetic>
	89	+```
	90	+
	91	+The ingress logger emits `update_id` and `run_id`; the scheduler and worker
	92	+queue decorators emit `message_id`, `tenant_id`, and `run_id`; the Telegram
	93	+delivery decorator emits `tenant_id`, `run_id`, `delivery_id`, and `chat_id`.
	94	+This preserves the same opaque correlation chain across process boundaries
	95	+without logging message bodies, credentials, or artifact contents.
	96	+
	97	+Worker invocation JSON and Go subtest durations are printed with `go test -v`.
	98	+These durations form the named local baseline for control-plane orchestration;
	99	+they intentionally exclude provider inference. Compare only runs using the same
	100	+commit, pinned images, host/runner class, cache state, and Docker architecture.
	101	+They are not a cloud cold-start or product latency qualification.
	102	+
	103	+## Operator inspection
	104	+
	105	+Use the opaque `tenant_id` and `run_id` printed by the test:
	106	+
	107	+```sql
	108	+SELECT status, context_epoch, created_at, updated_at
	109	+FROM runs
	110	+WHERE tenant_id = $tenant_id AND run_id = $run_id;
	111	+
	112	+SELECT attempt_number, status, worker_id, updated_at
	113	+FROM attempts
	114	+WHERE tenant_id = $tenant_id AND run_id = $run_id;
	115	+
	116	+SELECT sequence, created_at
	117	+FROM checkpoints
	118	+WHERE tenant_id = $tenant_id AND run_id = $run_id
	119	+ORDER BY sequence;
	120	+
	121	+SELECT source, input_tokens, output_tokens, observed_at
	122	+FROM usage_events
	123	+WHERE tenant_id = $tenant_id AND run_id = $run_id;
	124	+
	125	+SELECT status, next_attempt_at, updated_at, payload
	126	+FROM telegram_delivery_outbox
	127	+WHERE tenant_id = $tenant_id AND run_id = $run_id;
	128	+```
	129	+
	130	+`attempt_count` is contained in the typed JSON payload; the indexed columns
	131	+above remain optimized for delivery scheduling rather than ad hoc reporting.
	132	+
	133	+Queue state is visible at `http://localhost:9325`; MinIO is visible at
	134	+`http://localhost:9001`. `docker compose --project-name sessionless-dev logs
	135	+control-api reconciler worker-runtime telegram-sender` provides component
	136	+logs. Never use these local credentials or endpoints in cloud environments.

Minternal/deterministicharness/driver.go+20-12

	@@ -14,10 +14,11 @@ import (
14	)	14	)
15		15
16	type Config struct {	16	type Config struct {
17	- Turns uint64	17	+ Turns uint64
18	- Artifacts uint64	18	+ Artifacts uint64
19	- FailAtTurn uint64	19	+ FailBeforeFirstTurn bool
20	- RetryableFail bool	20	+ FailAtTurn uint64
		21	+ RetryableFail bool
21	}	22	}
22		23
23	type Driver struct {	24	type Driver struct {
	@@ -48,6 +49,9 @@ func (driver *Driver) Execute(
48	if sink == nil {	49	if sink == nil {
49	return ports.ExecutionResult{}, fmt.Errorf("execution event sink is required")	50	return ports.ExecutionResult{}, fmt.Errorf("execution event sink is required")
50	}	51	}
		52	+ if driver.config.FailBeforeFirstTurn && request.ResumeCheckpoint == nil {
		53	+ return ports.ExecutionResult{}, driver.failure()
		54	+ }
51	start := uint64(1)	55	start := uint64(1)
52	if request.ResumeCheckpoint != nil {	56	if request.ResumeCheckpoint != nil {
53	start = request.ResumeCheckpoint.Sequence + 1	57	start = request.ResumeCheckpoint.Sequence + 1
	@@ -71,14 +75,7 @@ func (driver *Driver) Execute(
71	return ports.ExecutionResult{}, err	75	return ports.ExecutionResult{}, err
72	}	76	}
73	if driver.config.FailAtTurn == turn {	77	if driver.config.FailAtTurn == turn {
74	- kind := domain.ErrorTerminal	78	+ return ports.ExecutionResult{}, driver.failure()
75	- if driver.config.RetryableFail {
76	- kind = domain.ErrorRetryable
77	- }
78	- return ports.ExecutionResult{}, &domain.ClassifiedError{
79	- Kind: kind, Code: "deterministic_failure",
80	- Operation: "deterministic_harness.execute",
81	- }
82	}	79	}
83	}	80	}
84	result := ports.ExecutionResult{	81	result := ports.ExecutionResult{
	@@ -102,6 +99,17 @@ func (driver *Driver) Execute(
102	return result, nil	99	return result, nil
103	}	100	}
104		101
		102	+func (driver *Driver) failure() error {
		103	+ kind := domain.ErrorTerminal
		104	+ if driver.config.RetryableFail {
		105	+ kind = domain.ErrorRetryable
		106	+ }
		107	+ return &domain.ClassifiedError{
		108	+ Kind: kind, Code: "deterministic_failure",
		109	+ Operation: "deterministic_harness.execute",
		110	+ }
		111	+}
		112	+
105	func (*Driver) Cancel(context.Context, ports.ExecutionIdentity) error {	113	func (*Driver) Cancel(context.Context, ports.ExecutionIdentity) error {
106	return nil	114	return nil
107	}	115	}

Ainternal/deterministicharness/driver_test.go+63-0文件内容审核中，请稍后刷新重试

Ainternal/portlog/ports.go+108-0文件内容审核中，请稍后刷新重试

Ainternal/portlog/ports_test.go+77-0文件内容审核中，请稍后刷新重试

Minternal/ports/ports.go+1-0

	@@ -169,6 +169,7 @@ type BlobStore interface {
169		169
170	type TelegramSendRequest struct {	170	type TelegramSendRequest struct {
171	TenantID domain.TenantID	171	TenantID domain.TenantID
		172	+ RunID domain.RunID
172	DeliveryID domain.TelegramDeliveryID	173	DeliveryID domain.TelegramDeliveryID
173	Chat domain.TelegramChatRef	174	Chat domain.TelegramChatRef
174	ReplyToMessageID int64	175	ReplyToMessageID int64

Minternal/ports/validation.go+3-0

	@@ -12,6 +12,9 @@ func (request TelegramSendRequest) Validate() error {
12	if err := request.TenantID.Validate(); err != nil {	12	if err := request.TenantID.Validate(); err != nil {
13	return err	13	return err
14	}	14	}
		15	+ if err := request.RunID.Validate(); err != nil {
		16	+ return err
		17	+ }
15	if err := request.DeliveryID.Validate(); err != nil {	18	if err := request.DeliveryID.Validate(); err != nil {
16	return err	19	return err
17	}	20	}

Minternal/ports/validation_test.go+2-0

	@@ -100,6 +100,7 @@ func TestTelegramSendRequestRejectsCrossTenantPayload(t *testing.T) {
100		100
101	request := ports.TelegramSendRequest{	101	request := ports.TelegramSendRequest{
102	TenantID: "tenant-a",	102	TenantID: "tenant-a",
		103	+ RunID: "run-1",
103	DeliveryID: "delivery-1",	104	DeliveryID: "delivery-1",
104	Chat: domain.TelegramChatRef{TenantID: "tenant-a", ChatID: -1000123},	105	Chat: domain.TelegramChatRef{TenantID: "tenant-a", ChatID: -1000123},
105	ReplyToMessageID: 77,	106	ReplyToMessageID: 77,
	@@ -116,6 +117,7 @@ func TestTelegramSendRequestAcceptsInlineTextOnly(t *testing.T) {
116		117
117	request := ports.TelegramSendRequest{	118	request := ports.TelegramSendRequest{
118	TenantID: "tenant-a",	119	TenantID: "tenant-a",
		120	+ RunID: "run-inline",
119	DeliveryID: "delivery-inline",	121	DeliveryID: "delivery-inline",
120	Chat: domain.TelegramChatRef{TenantID: "tenant-a", ChatID: 123},	122	Chat: domain.TelegramChatRef{TenantID: "tenant-a", ChatID: 123},
121	ReplyToMessageID: 78,	123	ReplyToMessageID: 78,

Minternal/telegramdelivery/client_test.go+1-1文件内容审核中，请稍后刷新重试

Minternal/telegramdelivery/sender.go+1-1文件内容审核中，请稍后刷新重试

Minternal/telegramfake/server.go+73-0

	@@ -41,6 +41,12 @@ type Server struct {
41	files map[string]storedFile	41	files map[string]storedFile
42	captures []Capture	42	captures []Capture
43	nextMessageID int64	43	nextMessageID int64
		44	+ failures map[string]failurePlan
		45	+}
		46	+
		47	+type failurePlan struct {
		48	+ Remaining int
		49	+ Status int
44	}	50	}
45		51
46	func NewHandler(token string, logger *slog.Logger) http.Handler {	52	func NewHandler(token string, logger *slog.Logger) http.Handler {
	@@ -50,6 +56,7 @@ func NewHandler(token string, logger *slog.Logger) http.Handler {
50	now: time.Now,	56	now: time.Now,
51	updates: make(map[int64]json.RawMessage),	57	updates: make(map[int64]json.RawMessage),
52	files: make(map[string]storedFile),	58	files: make(map[string]storedFile),
		59	+ failures: make(map[string]failurePlan),
53	nextMessageID: 1000,	60	nextMessageID: 1000,
54	}	61	}
55		62
	@@ -58,6 +65,7 @@ func NewHandler(token string, logger *slog.Logger) http.Handler {
58	mux.HandleFunc("POST /test/updates", server.addUpdate)	65	mux.HandleFunc("POST /test/updates", server.addUpdate)
59	mux.HandleFunc("POST /test/files/{fileID}", server.addFile)	66	mux.HandleFunc("POST /test/files/{fileID}", server.addFile)
60	mux.HandleFunc("GET /test/captures", server.listCaptures)	67	mux.HandleFunc("GET /test/captures", server.listCaptures)
		68	+ mux.HandleFunc("POST /test/failures", server.setFailure)
61	mux.HandleFunc("POST /test/reset", server.reset)	69	mux.HandleFunc("POST /test/reset", server.reset)
62	mux.HandleFunc("/", server.botAPI)	70	mux.HandleFunc("/", server.botAPI)
63	return mux	71	return mux
	@@ -109,11 +117,57 @@ func (server Server) reset(w http.ResponseWriter, _ http.Request) {
109	server.updateOrder = nil	117	server.updateOrder = nil
110	server.files = make(map[string]storedFile)	118	server.files = make(map[string]storedFile)
111	server.captures = nil	119	server.captures = nil
		120	+ server.failures = make(map[string]failurePlan)
112	server.nextMessageID = 1000	121	server.nextMessageID = 1000
113	server.mu.Unlock()	122	server.mu.Unlock()
114	writeTelegramResult(w, true)	123	writeTelegramResult(w, true)
115	}	124	}
116		125
		126	+func (server Server) setFailure(w http.ResponseWriter, request http.Request) {
		127	+ raw, err := readBody(request)
		128	+ if err != nil {
		129	+ writeError(w, http.StatusBadRequest, err)
		130	+ return
		131	+ }
		132	+ var plan struct {
		133	+ Method string `json:"method"`
		134	+ Count int `json:"count"`
		135	+ Status int `json:"status"`
		136	+ }
		137	+ if err := json.Unmarshal(raw, &plan); err != nil {
		138	+ writeError(w, http.StatusBadRequest, fmt.Errorf("decode failure plan: %w", err))
		139	+ return
		140	+ }
		141	+ switch plan.Method {
		142	+ case "sendMessage", "sendDocument", "sendPhoto":
		143	+ default:
		144	+ writeError(w, http.StatusBadRequest, fmt.Errorf("unsupported failure method"))
		145	+ return
		146	+ }
		147	+ if plan.Count < 0 {
		148	+ writeError(w, http.StatusBadRequest, fmt.Errorf("failure count must not be negative"))
		149	+ return
		150	+ }
		151	+ if plan.Status == 0 {
		152	+ plan.Status = http.StatusTooManyRequests
		153	+ }
		154	+ if plan.Status < 400 \|\| plan.Status > 599 {
		155	+ writeError(w, http.StatusBadRequest, fmt.Errorf("failure status must be 4xx or 5xx"))
		156	+ return
		157	+ }
		158	+ server.mu.Lock()
		159	+ if plan.Count == 0 {
		160	+ delete(server.failures, plan.Method)
		161	+ } else {
		162	+ server.failures[plan.Method] = failurePlan{
		163	+ Remaining: plan.Count,
		164	+ Status: plan.Status,
		165	+ }
		166	+ }
		167	+ server.mu.Unlock()
		168	+ writeTelegramResult(w, true)
		169	+}
		170	+
117	func (server Server) botAPI(w http.ResponseWriter, request http.Request) {	171	func (server Server) botAPI(w http.ResponseWriter, request http.Request) {
118	if request.Method != http.MethodPost && request.Method != http.MethodGet {	172	if request.Method != http.MethodPost && request.Method != http.MethodGet {
119	writeError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))	173	writeError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
	@@ -261,6 +315,25 @@ func (server Server) getUpdates(w http.ResponseWriter, request http.Request) {
261	}	315	}
262		316
263	func (server Server) captureSend(w http.ResponseWriter, request http.Request, method string) {	317	func (server Server) captureSend(w http.ResponseWriter, request http.Request, method string) {
		318	+ server.mu.Lock()
		319	+ plan := server.failures[method]
		320	+ if plan.Remaining > 0 {
		321	+ plan.Remaining--
		322	+ if plan.Remaining == 0 {
		323	+ delete(server.failures, method)
		324	+ } else {
		325	+ server.failures[method] = plan
		326	+ }
		327	+ server.mu.Unlock()
		328	+ writeJSON(w, plan.Status, map[string]any{
		329	+ "ok": false,
		330	+ "error_code": plan.Status,
		331	+ "description": "Injected deterministic Telegram failure",
		332	+ })
		333	+ return
		334	+ }
		335	+ server.mu.Unlock()
		336	+
264	raw, err := readRequestPayload(request)	337	raw, err := readRequestPayload(request)
265	if err != nil {	338	if err != nil {
266	writeError(w, http.StatusBadRequest, err)	339	writeError(w, http.StatusBadRequest, err)

Minternal/telegramfake/server_test.go+45-0

	@@ -59,6 +59,51 @@ func TestSendMessageIsCaptured(t *testing.T) {
59	}	59	}
60	}	60	}
61		61
		62	+func TestInjectedFailureIsBoundedAndNotCaptured(t *testing.T) {
		63	+ handler := NewHandler("test-token", slog.New(slog.NewTextHandler(io.Discard, nil)))
		64	+ response := performRequest(
		65	+ t,
		66	+ handler,
		67	+ http.MethodPost,
		68	+ "/test/failures",
		69	+ []byte(`{"method":"sendMessage","count":1,"status":429}`),
		70	+ )
		71	+ if response.Code != http.StatusOK {
		72	+ t.Fatalf("failure-plan status = %d, body = %s", response.Code, response.Body.String())
		73	+ }
		74	+
		75	+ first := performRequest(
		76	+ t,
		77	+ handler,
		78	+ http.MethodPost,
		79	+ "/bottest-token/sendMessage",
		80	+ []byte(`{"chat_id":123,"text":"retry me"}`),
		81	+ )
		82	+ if first.Code != http.StatusTooManyRequests {
		83	+ t.Fatalf("first send status = %d", first.Code)
		84	+ }
		85	+ second := performRequest(
		86	+ t,
		87	+ handler,
		88	+ http.MethodPost,
		89	+ "/bottest-token/sendMessage",
		90	+ []byte(`{"chat_id":123,"text":"retry me"}`),
		91	+ )
		92	+ if second.Code != http.StatusOK {
		93	+ t.Fatalf("second send status = %d, body = %s", second.Code, second.Body.String())
		94	+ }
		95	+
		96	+ response = performRequest(t, handler, http.MethodGet, "/test/captures", nil)
		97	+ var payload struct {
		98	+ OK bool `json:"ok"`
		99	+ Result []Capture `json:"result"`
		100	+ }
		101	+ decodeResponse(t, response, &payload)
		102	+ if !payload.OK \|\| len(payload.Result) != 1 {
		103	+ t.Fatalf("captures = %#v", payload)
		104	+ }
		105	+}
		106	+
62	func TestWrongBotTokenIsRejected(t *testing.T) {	107	func TestWrongBotTokenIsRejected(t *testing.T) {
63	handler := NewHandler("test-token", slog.New(slog.NewTextHandler(io.Discard, nil)))	108	handler := NewHandler("test-token", slog.New(slog.NewTextHandler(io.Discard, nil)))
64	response := performRequest(t, handler, http.MethodGet, "/botwrong/getMe", nil)	109	response := performRequest(t, handler, http.MethodGet, "/botwrong/getMe", nil)

Minternal/telegramingress/processor_test.go+6-0文件内容审核中，请稍后刷新重试

Ascripts/e2e-local.sh+29-0

@@ -0,0 +1,29 @@
	1	+#!/bin/sh
	2	+set -eu
	3	+
	4	+repo_root=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)
	5	+project_name=sessionless-dev
	6	+
	7	+cd "$repo_root"
	8	+
	9	+failure_logs() {
	10	+ status=$?
	11	+ if [ "$status" -ne 0 ]; then
	12	+ docker compose --project-name "$project_name" ps >&2 \|\| true
	13	+ docker compose --project-name "$project_name" logs \
	14	+ --no-color --tail 150 \
	15	+ control-api reconciler telegram-sender telegram-fake queue-local >&2 \|\| true
	16	+ fi
	17	+ exit "$status"
	18	+}
	19	+trap failure_logs EXIT HUP INT TERM
	20	+
	21	+make dev-up
	22	+docker compose --project-name "$project_name" --profile worker build worker-runtime
	23	+
	24	+YDB_CONNECTION_STRING=${YDB_CONNECTION_STRING:-grpc://localhost:${YDB_GRPC_PORT:-2136}/local?go_query_mode=scripting&go_fake_tx=scripting&go_query_bind=declare,numeric}
	25	+YDB_ANONYMOUS_CREDENTIALS=${YDB_ANONYMOUS_CREDENTIALS:-1}
	26	+SESSIONLESS_E2E=1
	27	+export YDB_CONNECTION_STRING YDB_ANONYMOUS_CREDENTIALS SESSIONLESS_E2E
	28	+
	29	+go test -v -count=1 -tags=e2elocal ./test/e2e/...

Atest/e2e/slice_test.go+846-0

@@ -0,0 +1,846 @@
	1	+//go:build e2elocal
	2	+
	3	+package e2e
	4	+
	5	+import (
	6	+ "bytes"
	7	+ "context"
	8	+ "database/sql"
	9	+ "encoding/json"
	10	+ "errors"
	11	+ "fmt"
	12	+ "io"
	13	+ "net/http"
	14	+ "os"
	15	+ "os/exec"
	16	+ "sort"
	17	+ "strconv"
	18	+ "strings"
	19	+ "testing"
	20	+ "time"
	21	+
	22	+ "gitcode.com/urandon/sessionless/internal/domain"
	23	+ "gitcode.com/urandon/sessionless/internal/queuecontract"
	24	+ "gitcode.com/urandon/sessionless/internal/s3store"
	25	+ "gitcode.com/urandon/sessionless/internal/sqsqueue"
	26	+ "gitcode.com/urandon/sessionless/internal/telegramingress"
	27	+ "gitcode.com/urandon/sessionless/internal/ydbclient"
	28	+ "gitcode.com/urandon/sessionless/internal/ydbstore"
	29	+)
	30	+
	31	+const (
	32	+ webhookSecret = "local-webhook-secret"
	33	+ identityKey = "sessionless-local-identity-key-0001"
	34	+ telegramSource = "bot-primary"
	35	+)
	36	+
	37	+type localSlice struct {
	38	+ t *testing.T
	39	+ ctx context.Context
	40	+ cancel context.CancelFunc
	41	+ db *sql.DB
	42	+ closeDB func()
	43	+ state *ydbstore.Store
	44	+ blobs *s3store.Store
	45	+ queue *sqsqueue.Queue
	46	+ client *http.Client
	47	+
	48	+ controlURL string
	49	+ telegramURL string
	50	+}
	51	+
	52	+type runRef struct {
	53	+ UpdateID int64
	54	+ MessageID int64
	55	+ ChatID int64
	56	+ TenantID domain.TenantID
	57	+ Conversation domain.ConversationRef
	58	+ ConnectionID domain.SubscriptionConnectionID
	59	+ RunID domain.RunID
	60	+}
	61	+
	62	+type capture struct {
	63	+ MessageID int64 `json:"message_id"`
	64	+ Method string `json:"method"`
	65	+ ChatID int64 `json:"chat_id"`
	66	+ Request json.RawMessage `json:"request"`
	67	+}
	68	+
	69	+func TestDeterministicLocalMultiUserSlice(t *testing.T) {
	70	+ if os.Getenv("SESSIONLESS_E2E") != "1" {
	71	+ t.Skip("set SESSIONLESS_E2E=1 and start the local stand")
	72	+ }
	73	+ slice := newLocalSlice(t)
	74	+ defer slice.close()
	75	+ slice.reset()
	76	+
	77	+ base := time.Now().UTC().UnixMilli()
	78	+ userA := int64(881_001)
	79	+ userB := int64(881_002)
	80	+
	81	+ t.Run("interleaved tenants, duplicate update, delivery retry and duplicate queue delivery", func(t *testing.T) {
	82	+ slice.injectTelegramFailure("sendMessage", 1, http.StatusTooManyRequests)
	83	+
	84	+ // Deliberately deliver the larger update first, then repeat it. Ordering
	85	+ // across users is irrelevant; idempotency is scoped to the Bot API update.
	86	+ runA := slice.postMessage(base+2, userA, "tenant A deterministic request")
	87	+ runB := slice.postMessageWithDocument(
	88	+ base+1, userB, "tenant B deterministic request with a file",
	89	+ "e2e-file-b", "notes.txt", []byte("synthetic tenant B attachment\n"),
	90	+ )
	91	+ duplicate := slice.postMessage(base+2, userA, "tenant A duplicate delivery")
	92	+ if duplicate.RunID != runA.RunID {
	93	+ t.Fatalf("duplicate update run = %s, want %s", duplicate.RunID, runA.RunID)
	94	+ }
	95	+
	96	+ slice.waitRunStatus(runA, domain.RunQuotaBlocked)
	97	+ slice.waitRunStatus(runB, domain.RunQuotaBlocked)
	98	+ slice.setConnectionReady(runA)
	99	+ slice.setConnectionReady(runB)
	100	+ slice.waitRunStatus(runA, domain.RunQueued)
	101	+ slice.waitRunStatus(runB, domain.RunQueued)
	102	+
	103	+ slice.runWorker(nil)
	104	+ slice.runWorker(nil)
	105	+ slice.waitRunStatus(runA, domain.RunSucceeded)
	106	+ slice.waitRunStatus(runB, domain.RunSucceeded)
	107	+ slice.waitForChatMethods(map[int64]map[string]int{
	108	+ userA: {"sendMessage": 1, "sendDocument": 1},
	109	+ userB: {"sendMessage": 1, "sendDocument": 1},
	110	+ })
	111	+
	112	+ slice.assertOneTelegramRun(runA)
	113	+ slice.assertOneTelegramRun(runB)
	114	+ slice.assertUsage(runA, 2)
	115	+ slice.assertUsage(runB, 2)
	116	+ slice.assertInputDocument(runB, "attachment-01-notes.txt")
	117	+ slice.assertTenantArtifacts(runA, runB)
	118	+ slice.assertDeliveryWasRetried(runA, runB)
	119	+
	120	+ before := len(slice.captures())
	121	+ slice.publishDuplicate(runA)
	122	+ slice.runWorker(nil)
	123	+ time.Sleep(750 * time.Millisecond)
	124	+ if after := len(slice.captures()); after != before {
	125	+ t.Fatalf("duplicate terminal delivery produced captures: before=%d after=%d", before, after)
	126	+ }
	127	+ })
	128	+
	129	+ t.Run("retry before the first checkpoint", func(t *testing.T) {
	130	+ run := slice.postMessage(base+10, userA, "retry before checkpoint")
	131	+ slice.setConnectionReady(run)
	132	+ slice.waitRunStatus(run, domain.RunQueued)
	133	+ slice.runWorker(map[string]string{
	134	+ "DETERMINISTIC_HARNESS_FAIL_BEFORE_FIRST_TURN": "true",
	135	+ "DETERMINISTIC_HARNESS_RETRYABLE_FAIL": "true",
	136	+ })
	137	+ slice.assertCheckpointCount(run, 0)
	138	+ time.Sleep(1200 * time.Millisecond)
	139	+ slice.runWorker(nil)
	140	+ slice.waitRunStatus(run, domain.RunSucceeded)
	141	+ slice.assertCheckpointCount(run, 2)
	142	+ })
	143	+
	144	+ t.Run("retry resumes after a durable checkpoint", func(t *testing.T) {
	145	+ run := slice.postMessage(base+11, userB, "retry after checkpoint")
	146	+ slice.setConnectionReady(run)
	147	+ slice.waitRunStatus(run, domain.RunQueued)
	148	+ slice.runWorker(map[string]string{
	149	+ "DETERMINISTIC_HARNESS_FAIL_AT_TURN": "1",
	150	+ "DETERMINISTIC_HARNESS_RETRYABLE_FAIL": "true",
	151	+ })
	152	+ slice.assertCheckpointCount(run, 1)
	153	+ time.Sleep(1200 * time.Millisecond)
	154	+ slice.runWorker(nil)
	155	+ slice.waitRunStatus(run, domain.RunSucceeded)
	156	+ slice.assertCheckpointCount(run, 2)
	157	+ })
	158	+
	159	+ t.Run("durable cancellation releases the run and replies in the same chat", func(t *testing.T) {
	160	+ run := slice.postMessage(base+12, userA, "cancel this run")
	161	+ slice.setConnectionReady(run)
	162	+ slice.waitRunStatus(run, domain.RunQueued)
	163	+ before := len(slice.capturesForChat(userA))
	164	+ slice.requestCancellation(run)
	165	+ slice.runWorker(nil)
	166	+ slice.waitRunStatus(run, domain.RunCancelled)
	167	+ slice.waitCaptureIncrease(userA, before)
	168	+ })
	169	+
	170	+ t.Run("provider quota block recovers without API billing fallback", func(t *testing.T) {
	171	+ run := slice.postMessage(base+13, userB, "wait for provider reset")
	172	+ resetAt := time.Now().UTC().Add(10 * time.Minute)
	173	+ slice.setConnectionState(
	174	+ run,
	175	+ domain.EntitlementActive,
	176	+ domain.ProviderQuotaExhausted,
	177	+ domain.SchedulerBlockedUntilReset,
	178	+ &resetAt,
	179	+ )
	180	+ slice.waitRunStatus(run, domain.RunQuotaBlocked)
	181	+ slice.setConnectionReady(run)
	182	+ slice.waitRunStatus(run, domain.RunQueued)
	183	+ slice.runWorker(nil)
	184	+ slice.waitRunStatus(run, domain.RunSucceeded)
	185	+ })
	186	+
	187	+ t.Run("admitted dispatch is republished after a queue outage", func(t *testing.T) {
	188	+ slice.compose("stop", "queue-local")
	189	+ run := slice.postMessage(base+14, userA, "repair dispatch publication gap")
	190	+ slice.setConnectionReady(run)
	191	+ slice.waitRunStatus(run, domain.RunQueued)
	192	+ slice.compose("start", "queue-local")
	193	+ slice.waitHTTP("http://127.0.0.1:9324/?Action=ListQueues&Version=2012-11-05")
	194	+ time.Sleep(500 * time.Millisecond)
	195	+ slice.runWorker(nil)
	196	+ slice.waitRunStatus(run, domain.RunSucceeded)
	197	+ })
	198	+
	199	+ t.Run("explicit clean context advances the frontend context epoch", func(t *testing.T) {
	200	+ before := len(slice.capturesForChat(userA))
	201	+ command := slice.postMessage(base+15, userA, "/new")
	202	+ slice.waitRunStatus(command, domain.RunSucceeded)
	203	+ slice.waitCaptureIncrease(userA, before)
	204	+ var epoch uint64
	205	+ if err := slice.db.QueryRowContext(
	206	+ slice.ctx,
	207	+ `SELECT current_context_epoch FROM conversations
	208	+ WHERE tenant_id = $1 AND conversation_id = $2`,
	209	+ command.TenantID, command.Conversation.ID,
	210	+ ).Scan(&epoch); err != nil {
	211	+ t.Fatal(err)
	212	+ }
	213	+ if epoch == 0 {
	214	+ t.Fatal("explicit clean context did not advance the context epoch")
	215	+ }
	216	+ })
	217	+}
	218	+
	219	+func newLocalSlice(t testing.T) localSlice {
	220	+ t.Helper()
	221	+ ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
	222	+ ydb, err := ydbclient.Open(ctx, envOrDefault(
	223	+ "YDB_CONNECTION_STRING",
	224	+ "grpc://localhost:2136/local?go_query_mode=scripting&go_fake_tx=scripting&go_query_bind=declare,numeric",
	225	+ ))
	226	+ if err != nil {
	227	+ cancel()
	228	+ t.Fatal(err)
	229	+ }
	230	+ state, err := ydbstore.New(ydb.DB, ydbstore.Options{})
	231	+ if err != nil {
	232	+ _ = ydb.Close(context.Background())
	233	+ cancel()
	234	+ t.Fatal(err)
	235	+ }
	236	+ blobs, err := s3store.New(ctx, s3store.Config{
	237	+ Endpoint: envOrDefault("S3_ENDPOINT", "http://localhost:9000"),
	238	+ Region: envOrDefault("S3_REGION", "us-east-1"),
	239	+ Bucket: envOrDefault("S3_BUCKET", "sessionless-local"),
	240	+ AccessKeyID: envOrDefault("S3_ACCESS_KEY_ID", "sessionless-local"),
	241	+ SecretAccessKey: envOrDefault("S3_SECRET_ACCESS_KEY", "sessionless-local-secret"),
	242	+ ForcePathStyle: true,
	243	+ })
	244	+ if err != nil {
	245	+ _ = ydb.Close(context.Background())
	246	+ cancel()
	247	+ t.Fatal(err)
	248	+ }
	249	+ queue, err := sqsqueue.New(ctx, sqsqueue.Config{
	250	+ Endpoint: envOrDefault("QUEUE_ENDPOINT", "http://localhost:9324"),
	251	+ Region: envOrDefault("QUEUE_REGION", "us-east-1"),
	252	+ QueueURL: envOrDefault("DISPATCH_QUEUE_URL", "http://localhost:9324/000000000000/sessionless-dispatch"),
	253	+ DeadLetterURL: envOrDefault("DEAD_LETTER_QUEUE_URL", "http://localhost:9324/000000000000/sessionless-dlq"),
	254	+ AccessKeyID: envOrDefault("QUEUE_ACCESS_KEY_ID", "sessionless-local"),
	255	+ SecretAccessKey: envOrDefault("QUEUE_SECRET_ACCESS_KEY", "sessionless-local-secret"),
	256	+ })
	257	+ if err != nil {
	258	+ _ = ydb.Close(context.Background())
	259	+ cancel()
	260	+ t.Fatal(err)
	261	+ }
	262	+ return &localSlice{
	263	+ t: t, ctx: ctx, cancel: cancel, db: ydb.DB,
	264	+ closeDB: func() { _ = ydb.Close(context.Background()) },
	265	+ state: state, blobs: blobs, queue: queue,
	266	+ client: &http.Client{Timeout: 10 * time.Second},
	267	+ controlURL: envOrDefault("SESSIONLESS_BASE_URL", "http://localhost:8080"),
	268	+ telegramURL: envOrDefault("TELEGRAM_API_BASE_URL", "http://localhost:8081"),
	269	+ }
	270	+}
	271	+
	272	+func (slice *localSlice) close() {
	273	+ slice.closeDB()
	274	+ slice.cancel()
	275	+}
	276	+
	277	+func (slice *localSlice) reset() {
	278	+ slice.postJSON(slice.telegramURL+"/test/reset", []byte(`{}`))
	279	+ for {
	280	+ message, err := slice.queue.Receive(slice.ctx)
	281	+ if errors.Is(err, sqsqueue.ErrNoMessage) {
	282	+ break
	283	+ }
	284	+ if err != nil {
	285	+ slice.t.Fatal(err)
	286	+ }
	287	+ if err := slice.queue.Ack(slice.ctx, message.ReceiptHandle); err != nil {
	288	+ slice.t.Fatal(err)
	289	+ }
	290	+ }
	291	+}
	292	+
	293	+func (slice *localSlice) postMessage(updateID, chatID int64, text string) runRef {
	294	+ slice.t.Helper()
	295	+ return slice.postUpdate(updateID, chatID, text, nil)
	296	+}
	297	+
	298	+func (slice *localSlice) postMessageWithDocument(
	299	+ updateID, chatID int64,
	300	+ text, fileID, name string,
	301	+ data []byte,
	302	+) runRef {
	303	+ slice.t.Helper()
	304	+ request, err := http.NewRequest(
	305	+ http.MethodPost,
	306	+ slice.telegramURL+"/test/files/"+fileID+"?name="+name,
	307	+ bytes.NewReader(data),
	308	+ )
	309	+ if err != nil {
	310	+ slice.t.Fatal(err)
	311	+ }
	312	+ request.Header.Set("Content-Type", "text/plain")
	313	+ response, err := slice.client.Do(request)
	314	+ if err != nil {
	315	+ slice.t.Fatal(err)
	316	+ }
	317	+ response.Body.Close()
	318	+ if response.StatusCode != http.StatusOK {
	319	+ slice.t.Fatalf("seed Telegram file status = %d", response.StatusCode)
	320	+ }
	321	+ return slice.postUpdate(updateID, chatID, text, map[string]any{
	322	+ "file_id": fileID, "file_name": name, "mime_type": "text/plain",
	323	+ })
	324	+}
	325	+
	326	+func (slice *localSlice) postUpdate(
	327	+ updateID, chatID int64,
	328	+ text string,
	329	+ document map[string]any,
	330	+) runRef {
	331	+ slice.t.Helper()
	332	+ message := map[string]any{
	333	+ "message_id": updateID,
	334	+ "from": map[string]any{"id": chatID},
	335	+ "chat": map[string]any{"id": chatID, "type": "private"},
	336	+ "date": time.Now().UTC().Unix(),
	337	+ "text": text,
	338	+ }
	339	+ if document != nil {
	340	+ message["document"] = document
	341	+ }
	342	+ body, err := json.Marshal(map[string]any{
	343	+ "update_id": updateID,
	344	+ "message": message,
	345	+ })
	346	+ if err != nil {
	347	+ slice.t.Fatal(err)
	348	+ }
	349	+ request, err := http.NewRequest(
	350	+ http.MethodPost, slice.controlURL+"/telegram/webhook", bytes.NewReader(body),
	351	+ )
	352	+ if err != nil {
	353	+ slice.t.Fatal(err)
	354	+ }
	355	+ request.Header.Set("Content-Type", "application/json")
	356	+ request.Header.Set("X-Telegram-Bot-Api-Secret-Token", webhookSecret)
	357	+ response, err := slice.client.Do(request)
	358	+ if err != nil {
	359	+ slice.t.Fatal(err)
	360	+ }
	361	+ response.Body.Close()
	362	+ if response.StatusCode != http.StatusOK {
	363	+ slice.t.Fatalf("webhook status = %d", response.StatusCode)
	364	+ }
	365	+
	366	+ resolver, err := telegramingress.NewIdentityResolver([]byte(identityKey))
	367	+ if err != nil {
	368	+ slice.t.Fatal(err)
	369	+ }
	370	+ identity, err := resolver.ResolvePrivate(chatID, chatID, "codex")
	371	+ if err != nil {
	372	+ slice.t.Fatal(err)
	373	+ }
	374	+ var runID domain.RunID
	375	+ if err := slice.db.QueryRowContext(
	376	+ slice.ctx,
	377	+ `SELECT run_id FROM telegram_updates
	378	+ WHERE tenant_id = $1 AND source_id = $2 AND update_id = $3`,
	379	+ identity.Tenant, telegramSource, updateID,
	380	+ ).Scan(&runID); err != nil {
	381	+ slice.t.Fatal(err)
	382	+ }
	383	+ ref := runRef{
	384	+ UpdateID: updateID, MessageID: updateID, ChatID: chatID,
	385	+ TenantID: identity.Tenant, Conversation: identity.Conversation,
	386	+ ConnectionID: identity.SubscriptionConnection, RunID: runID,
	387	+ }
	388	+ slice.t.Logf(
	389	+ "correlation update_id=%d tenant_id=%s run_id=%s chat_id=%d",
	390	+ ref.UpdateID, ref.TenantID, ref.RunID, ref.ChatID,
	391	+ )
	392	+ return ref
	393	+}
	394	+
	395	+func (slice *localSlice) assertInputDocument(run runRef, name string) {
	396	+ slice.t.Helper()
	397	+ loaded, found, err := slice.state.LoadWorkerJob(slice.ctx, run.TenantID, run.RunID)
	398	+ if err != nil {
	399	+ slice.t.Fatal(err)
	400	+ }
	401	+ if !found {
	402	+ slice.t.Fatalf("worker job for run %s not found", run.RunID)
	403	+ }
	404	+ for _, artifact := range loaded.InputManifest.Artifacts {
	405	+ if artifact.Name == name {
	406	+ if !strings.HasPrefix(artifact.Blob.Key, domain.TenantObjectPrefix(run.TenantID)) {
	407	+ slice.t.Fatalf("input document key = %q", artifact.Blob.Key)
	408	+ }
	409	+ return
	410	+ }
	411	+ }
	412	+ slice.t.Fatalf("input document %q not found in worker manifest", name)
	413	+}
	414	+
	415	+func (slice *localSlice) setConnectionReady(run runRef) {
	416	+ slice.setConnectionState(
	417	+ run,
	418	+ domain.EntitlementActive,
	419	+ domain.ProviderQuotaUnknown,
	420	+ domain.SchedulerReady,
	421	+ nil,
	422	+ )
	423	+}
	424	+
	425	+func (slice *localSlice) setConnectionState(
	426	+ run runRef,
	427	+ entitlement domain.EntitlementState,
	428	+ quota domain.ProviderQuotaState,
	429	+ scheduler domain.SchedulerState,
	430	+ blockedUntil *time.Time,
	431	+) {
	432	+ slice.t.Helper()
	433	+ blocked := time.Unix(0, 0).UTC()
	434	+ if blockedUntil != nil {
	435	+ blocked = blockedUntil.UTC()
	436	+ }
	437	+ if _, err := slice.db.ExecContext(
	438	+ slice.ctx,
	439	+ `UPDATE subscription_connections
	440	+ SET entitlement_state = $1, quota_state = $2, updated_at = CurrentUtcTimestamp()
	441	+ WHERE tenant_id = $3 AND subscription_connection_id = $4`,
	442	+ entitlement, quota, run.TenantID, run.ConnectionID,
	443	+ ); err != nil {
	444	+ slice.t.Fatal(err)
	445	+ }
	446	+ if _, err := slice.db.ExecContext(
	447	+ slice.ctx,
	448	+ `UPDATE subscription_scheduler_slots
	449	+ SET state = $1, blocked_until = $2, updated_at = CurrentUtcTimestamp()
	450	+ WHERE tenant_id = $3 AND subscription_connection_id = $4`,
	451	+ scheduler, blocked, run.TenantID, run.ConnectionID,
	452	+ ); err != nil {
	453	+ slice.t.Fatal(err)
	454	+ }
	455	+}
	456	+
	457	+func (slice *localSlice) waitRunStatus(run runRef, wanted domain.RunStatus) {
	458	+ slice.t.Helper()
	459	+ deadline := time.Now().Add(35 * time.Second)
	460	+ for {
	461	+ var status domain.RunStatus
	462	+ err := slice.db.QueryRowContext(
	463	+ slice.ctx,
	464	+ `SELECT status FROM runs WHERE tenant_id = $1 AND run_id = $2`,
	465	+ run.TenantID, run.RunID,
	466	+ ).Scan(&status)
	467	+ if err == nil && status == wanted {
	468	+ return
	469	+ }
	470	+ if time.Now().After(deadline) {
	471	+ slice.t.Fatalf("run %s status = %q, want %q (last error: %v)", run.RunID, status, wanted, err)
	472	+ }
	473	+ time.Sleep(100 * time.Millisecond)
	474	+ }
	475	+}
	476	+
	477	+func (slice *localSlice) runWorker(overrides map[string]string) {
	478	+ slice.t.Helper()
	479	+ arguments := []string{
	480	+ "compose", "--project-name", "sessionless-dev", "--profile", "worker",
	481	+ "run", "--rm", "--no-deps",
	482	+ "-e", "WORKER_RETRY_DELAY=1s",
	483	+ "-e", "WORKER_QUEUE_WAIT=2s",
	484	+ }
	485	+ keys := make([]string, 0, len(overrides))
	486	+ for key := range overrides {
	487	+ keys = append(keys, key)
	488	+ }
	489	+ sort.Strings(keys)
	490	+ for _, key := range keys {
	491	+ arguments = append(arguments, "-e", key+"="+overrides[key])
	492	+ }
	493	+ arguments = append(arguments, "worker-runtime")
	494	+ command := exec.CommandContext(slice.ctx, "docker", arguments...)
	495	+ command.Dir = repositoryRoot(slice.t)
	496	+ output, err := command.CombinedOutput()
	497	+ if err != nil {
	498	+ slice.t.Fatalf("worker container failed: %v\n%s", err, output)
	499	+ }
	500	+ slice.t.Logf("worker invocation: %s", strings.TrimSpace(string(output)))
	501	+}
	502	+
	503	+func (slice *localSlice) publishDuplicate(run runRef) {
	504	+ slice.t.Helper()
	505	+ if err := slice.queue.Publish(slice.ctx, queuecontract.Envelope{
	506	+ Schema: queuecontract.SchemaV1,
	507	+ MessageID: domain.MessageID(
	508	+ "msg-e2e-duplicate-" + strconv.FormatInt(run.UpdateID, 10),
	509	+ ),
	510	+ Kind: queuecontract.KindDispatchRun, TenantID: run.TenantID,
	511	+ SubjectID: string(run.RunID), EnqueuedAt: time.Now().UTC(),
	512	+ }); err != nil {
	513	+ slice.t.Fatal(err)
	514	+ }
	515	+}
	516	+
	517	+func (slice *localSlice) requestCancellation(ref runRef) {
	518	+ slice.t.Helper()
	519	+ var payload string
	520	+ if err := slice.db.QueryRowContext(
	521	+ slice.ctx,
	522	+ `SELECT payload FROM runs WHERE tenant_id = $1 AND run_id = $2`,
	523	+ ref.TenantID, ref.RunID,
	524	+ ).Scan(&payload); err != nil {
	525	+ slice.t.Fatal(err)
	526	+ }
	527	+ var run domain.Run
	528	+ if err := json.Unmarshal([]byte(payload), &run); err != nil {
	529	+ slice.t.Fatal(err)
	530	+ }
	531	+ now := time.Now().UTC()
	532	+ run.CancellationRequestedAt = &now
	533	+ run.UpdatedAt = now
	534	+ updated, err := json.Marshal(run)
	535	+ if err != nil {
	536	+ slice.t.Fatal(err)
	537	+ }
	538	+ if _, err := slice.db.ExecContext(
	539	+ slice.ctx,
	540	+ `UPDATE runs
	541	+ SET updated_at = $1, payload = CAST($2 AS JsonDocument)
	542	+ WHERE tenant_id = $3 AND run_id = $4`,
	543	+ now, string(updated), ref.TenantID, ref.RunID,
	544	+ ); err != nil {
	545	+ slice.t.Fatal(err)
	546	+ }
	547	+}
	548	+
	549	+func (slice *localSlice) assertOneTelegramRun(run runRef) {
	550	+ slice.t.Helper()
	551	+ var count uint64
	552	+ if err := slice.db.QueryRowContext(
	553	+ slice.ctx,
	554	+ `SELECT COUNT(*) FROM telegram_updates
	555	+ WHERE tenant_id = $1 AND source_id = $2 AND update_id = $3`,
	556	+ run.TenantID, telegramSource, run.UpdateID,
	557	+ ).Scan(&count); err != nil {
	558	+ slice.t.Fatal(err)
	559	+ }
	560	+ if count != 1 {
	561	+ slice.t.Fatalf("update %d rows = %d, want 1", run.UpdateID, count)
	562	+ }
	563	+}
	564	+
	565	+func (slice *localSlice) assertCheckpointCount(run runRef, wanted uint64) {
	566	+ slice.t.Helper()
	567	+ var count uint64
	568	+ if err := slice.db.QueryRowContext(
	569	+ slice.ctx,
	570	+ `SELECT COUNT(*) FROM checkpoints WHERE tenant_id = $1 AND run_id = $2`,
	571	+ run.TenantID, run.RunID,
	572	+ ).Scan(&count); err != nil {
	573	+ slice.t.Fatal(err)
	574	+ }
	575	+ if count != wanted {
	576	+ slice.t.Fatalf("run %s checkpoints = %d, want %d", run.RunID, count, wanted)
	577	+ }
	578	+}
	579	+
	580	+func (slice *localSlice) assertUsage(run runRef, wanted uint64) {
	581	+ slice.t.Helper()
	582	+ var count uint64
	583	+ if err := slice.db.QueryRowContext(
	584	+ slice.ctx,
	585	+ `SELECT COUNT(*) FROM usage_observations WHERE tenant_id = $1 AND run_id = $2`,
	586	+ run.TenantID, run.RunID,
	587	+ ).Scan(&count); err != nil {
	588	+ slice.t.Fatal(err)
	589	+ }
	590	+ if count != wanted {
	591	+ slice.t.Fatalf("run %s usage rows = %d, want %d", run.RunID, count, wanted)
	592	+ }
	593	+}
	594	+
	595	+func (slice *localSlice) assertTenantArtifacts(runA, runB runRef) {
	596	+ slice.t.Helper()
	597	+ manifestA := slice.outputManifest(runA)
	598	+ manifestB := slice.outputManifest(runB)
	599	+ if len(manifestA.Artifacts) == 0 \|\| len(manifestB.Artifacts) == 0 {
	600	+ slice.t.Fatal("terminal output manifests must contain artifacts")
	601	+ }
	602	+ for _, item := range manifestA.Artifacts {
	603	+ if !strings.HasPrefix(item.Blob.Key, domain.TenantObjectPrefix(runA.TenantID)) {
	604	+ slice.t.Fatalf("tenant A artifact key = %q", item.Blob.Key)
	605	+ }
	606	+ }
	607	+ for _, item := range manifestB.Artifacts {
	608	+ if !strings.HasPrefix(item.Blob.Key, domain.TenantObjectPrefix(runB.TenantID)) {
	609	+ slice.t.Fatalf("tenant B artifact key = %q", item.Blob.Key)
	610	+ }
	611	+ }
	612	+ _, err := slice.blobs.Open(slice.ctx, runB.TenantID, manifestA.Artifacts[0].Blob)
	613	+ var mismatch domain.TenantMismatchError
	614	+ if !errors.As(err, &mismatch) {
	615	+ slice.t.Fatalf("cross-tenant artifact open error = %v, want TenantMismatchError", err)
	616	+ }
	617	+}
	618	+
	619	+func (slice *localSlice) outputManifest(run runRef) domain.ArtifactManifest {
	620	+ slice.t.Helper()
	621	+ delivery := slice.deliveryForRun(run)
	622	+ if delivery.ArtifactManifestID == nil {
	623	+ slice.t.Fatalf("run %s has no terminal artifact manifest", run.RunID)
	624	+ }
	625	+ manifest, found, err := slice.state.GetArtifactManifest(
	626	+ slice.ctx, run.TenantID, *delivery.ArtifactManifestID,
	627	+ )
	628	+ if err != nil {
	629	+ slice.t.Fatal(err)
	630	+ }
	631	+ if !found {
	632	+ slice.t.Fatalf("manifest %s not found", *delivery.ArtifactManifestID)
	633	+ }
	634	+ return manifest
	635	+}
	636	+
	637	+func (slice *localSlice) assertDeliveryWasRetried(runs ...runRef) {
	638	+ slice.t.Helper()
	639	+ for _, run := range runs {
	640	+ if slice.deliveryForRun(run).AttemptCount >= 2 {
	641	+ return
	642	+ }
	643	+ }
	644	+ slice.t.Fatal("injected Telegram failure did not produce a durable retry")
	645	+}
	646	+
	647	+func (slice *localSlice) deliveryForRun(run runRef) domain.TelegramDeliveryOutbox {
	648	+ slice.t.Helper()
	649	+ var payload string
	650	+ if err := slice.db.QueryRowContext(
	651	+ slice.ctx,
	652	+ `SELECT payload FROM telegram_delivery_outbox
	653	+ WHERE tenant_id = $1 AND run_id = $2 LIMIT 1`,
	654	+ run.TenantID, run.RunID,
	655	+ ).Scan(&payload); err != nil {
	656	+ slice.t.Fatal(err)
	657	+ }
	658	+ var delivery domain.TelegramDeliveryOutbox
	659	+ if err := json.Unmarshal([]byte(payload), &delivery); err != nil {
	660	+ slice.t.Fatal(err)
	661	+ }
	662	+ return delivery
	663	+}
	664	+
	665	+func (slice *localSlice) injectTelegramFailure(method string, count, status int) {
	666	+ slice.t.Helper()
	667	+ body, err := json.Marshal(map[string]any{
	668	+ "method": method, "count": count, "status": status,
	669	+ })
	670	+ if err != nil {
	671	+ slice.t.Fatal(err)
	672	+ }
	673	+ slice.postJSON(slice.telegramURL+"/test/failures", body)
	674	+}
	675	+
	676	+func (slice *localSlice) waitForChatMethods(wanted map[int64]map[string]int) {
	677	+ slice.t.Helper()
	678	+ deadline := time.Now().Add(35 * time.Second)
	679	+ for {
	680	+ got := make(map[int64]map[string]int)
	681	+ for _, item := range slice.captures() {
	682	+ if got[item.ChatID] == nil {
	683	+ got[item.ChatID] = make(map[string]int)
	684	+ }
	685	+ got[item.ChatID][item.Method]++
	686	+ }
	687	+ complete := true
	688	+ for chatID, methods := range wanted {
	689	+ for method, count := range methods {
	690	+ if got[chatID][method] < count {
	691	+ complete = false
	692	+ }
	693	+ }
	694	+ }
	695	+ if complete {
	696	+ return
	697	+ }
	698	+ if time.Now().After(deadline) {
	699	+ slice.t.Fatalf("Telegram captures = %#v, want %#v", got, wanted)
	700	+ }
	701	+ time.Sleep(100 * time.Millisecond)
	702	+ }
	703	+}
	704	+
	705	+func (slice *localSlice) waitCaptureIncrease(chatID int64, before int) {
	706	+ slice.t.Helper()
	707	+ deadline := time.Now().Add(35 * time.Second)
	708	+ for {
	709	+ if len(slice.capturesForChat(chatID)) > before {
	710	+ return
	711	+ }
	712	+ if time.Now().After(deadline) {
	713	+ slice.t.Fatalf("no new Telegram capture for chat %d", chatID)
	714	+ }
	715	+ time.Sleep(100 * time.Millisecond)
	716	+ }
	717	+}
	718	+
	719	+func (slice *localSlice) capturesForChat(chatID int64) []capture {
	720	+ all := slice.captures()
	721	+ result := make([]capture, 0)
	722	+ for _, item := range all {
	723	+ if item.ChatID == chatID {
	724	+ result = append(result, item)
	725	+ }
	726	+ }
	727	+ return result
	728	+}
	729	+
	730	+func (slice *localSlice) captures() []capture {
	731	+ slice.t.Helper()
	732	+ var payload struct {
	733	+ OK bool `json:"ok"`
	734	+ Result []capture `json:"result"`
	735	+ }
	736	+ slice.getJSON(slice.telegramURL+"/test/captures", &payload)
	737	+ if !payload.OK {
	738	+ slice.t.Fatal("Telegram fake returned ok=false")
	739	+ }
	740	+ return payload.Result
	741	+}
	742	+
	743	+func (slice *localSlice) compose(arguments ...string) {
	744	+ slice.t.Helper()
	745	+ args := append([]string{"compose", "--project-name", "sessionless-dev"}, arguments...)
	746	+ command := exec.CommandContext(slice.ctx, "docker", args...)
	747	+ command.Dir = repositoryRoot(slice.t)
	748	+ output, err := command.CombinedOutput()
	749	+ if err != nil {
	750	+ slice.t.Fatalf("docker compose %s: %v\n%s", strings.Join(arguments, " "), err, output)
	751	+ }
	752	+}
	753	+
	754	+func (slice *localSlice) waitHTTP(url string) {
	755	+ slice.t.Helper()
	756	+ deadline := time.Now().Add(30 * time.Second)
	757	+ for {
	758	+ response, err := slice.client.Get(url)
	759	+ if err == nil {
	760	+ response.Body.Close()
	761	+ if response.StatusCode >= 200 && response.StatusCode < 500 {
	762	+ return
	763	+ }
	764	+ }
	765	+ if time.Now().After(deadline) {
	766	+ slice.t.Fatalf("%s did not become ready: %v", url, err)
	767	+ }
	768	+ time.Sleep(200 * time.Millisecond)
	769	+ }
	770	+}
	771	+
	772	+func (slice *localSlice) postJSON(url string, body []byte) {
	773	+ slice.t.Helper()
	774	+ request, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
	775	+ if err != nil {
	776	+ slice.t.Fatal(err)
	777	+ }
	778	+ request.Header.Set("Content-Type", "application/json")
	779	+ response, err := slice.client.Do(request)
	780	+ if err != nil {
	781	+ slice.t.Fatal(err)
	782	+ }
	783	+ defer response.Body.Close()
	784	+ if response.StatusCode != http.StatusOK {
	785	+ data, _ := io.ReadAll(response.Body)
	786	+ slice.t.Fatalf("POST %s: status=%d body=%s", url, response.StatusCode, data)
	787	+ }
	788	+}
	789	+
	790	+func (slice *localSlice) getJSON(url string, target any) {
	791	+ slice.t.Helper()
	792	+ response, err := slice.client.Get(url)
	793	+ if err != nil {
	794	+ slice.t.Fatal(err)
	795	+ }
	796	+ defer response.Body.Close()
	797	+ if response.StatusCode != http.StatusOK {
	798	+ data, _ := io.ReadAll(response.Body)
	799	+ slice.t.Fatalf("GET %s: status=%d body=%s", url, response.StatusCode, data)
	800	+ }
	801	+ if err := json.NewDecoder(response.Body).Decode(target); err != nil {
	802	+ slice.t.Fatal(err)
	803	+ }
	804	+}
	805	+
	806	+func repositoryRoot(t *testing.T) string {
	807	+ t.Helper()
	808	+ directory, err := os.Getwd()
	809	+ if err != nil {
	810	+ t.Fatal(err)
	811	+ }
	812	+ for {
	813	+ if _, err := os.Stat(directory + "/go.mod"); err == nil {
	814	+ return directory
	815	+ }
	816	+ parent := directory[:strings.LastIndex(directory, "/")]
	817	+ if parent == "" \|\| parent == directory {
	818	+ t.Fatal("repository root not found")
	819	+ }
	820	+ directory = parent
	821	+ }
	822	+}
	823	+
	824	+func envOrDefault(name, fallback string) string {
	825	+ if value := os.Getenv(name); value != "" {
	826	+ return value
	827	+ }
	828	+ return fallback
	829	+}
	830	+
	831	+func TestE2EPackageCompilesWithoutRuntimeEnvironment(t *testing.T) {
	832	+ if os.Getenv("SESSIONLESS_E2E") == "1" {
	833	+ return
	834	+ }
	835	+ // This test deliberately exercises only pure helpers so the tagged package
	836	+ // can be compiled in a preflight job without Docker or live adapters.
	837	+ if root := repositoryRoot(t); root == "" {
	838	+ t.Fatal("repository root is empty")
	839	+ }
	840	+ if _, err := strconv.ParseInt("42", 10, 64); err != nil {
	841	+ t.Fatal(err)
	842	+ }
	843	+ if fmt.Sprint(domain.RunSucceeded) != "succeeded" {
	844	+ t.Fatal("unexpected domain status")
	845	+ }
	846	+}

Mtest/fixtures/README.md+5-0文件内容审核中，请稍后刷新重试