Require dedupe timestamp in job queue payloads + dedupe

Author: parsley42

aidocs/JobQueues.md

@@ -73,11 +73,12 @@ Identity model:
 - Parse queue bodies of the form:
 
 ```text
-<uuid> <shell-escaped arguments>
+<uuid>:<timestamp> <shell-escaped arguments>
 ```
 
-- Match only the first UUID-width bytes of the body against configured job
-  UUIDs.
+- Match the UUID prefix against configured job UUIDs.
+- Require a 12-15 digit numeric timestamp after the UUID and colon.
+- Deduplicate matching queue triggers by `<uuid>:<timestamp>` for 140 seconds.
 - Preserve job arguments with spaces by parsing shell-escaped argument text.
 - Log an error for malformed or unknown UUID bodies, and log an info event when
   a job is triggered from a queue.
@@ -272,9 +273,12 @@ Rules:
 - Bodies shorter than 36 bytes are invalid.
 - Only bytes `[0:36]` are considered for job selection.
 - The UUID prefix must parse as a UUID.
-- UUID-only payloads are valid and mean the job has zero arguments.
-- If the body has more data, byte 36 must be a single ASCII space.
-- Argument text begins at byte 37.
+- Byte 36 must be a single ASCII colon.
+- A numeric timestamp of 12-15 digits must follow the colon.
+- Timestamp-only payloads are valid and mean the job has zero arguments.
+- If the body has more data after the timestamp, the next byte must be a single
+  ASCII space.
+- Argument text begins after that space.
 - Empty argument text means zero arguments.
 - The argument parser tokenizes shell-escaped words without command execution,
   variable expansion, globbing, or file access.
@@ -286,13 +290,27 @@ Rules:
 Examples:
 
 ```text
-1104df4c-feeb-43ab-8c85-83663288cea9 alpha two\ words
-1104df4c-feeb-43ab-8c85-83663288cea9 'alpha beta' gamma
+1104df4c-feeb-43ab-8c85-83663288cea9:17642656976077
+1104df4c-feeb-43ab-8c85-83663288cea9:17642656976077 alpha two\ words
+1104df4c-feeb-43ab-8c85-83663288cea9:17642656976077 'alpha beta' gamma
 ```
 
 The existing `resources/gcloud/scripts/queue-job.sh` POC matches this contract
-by using `printf "%q "` for each argument and sending the payload as
-`text/plain`.
+by calculating the timestamp with `echo "$(( $(date +%s%N) / 100000 ))"`,
+using `printf "%q "` for each argument, and sending the payload as `text/plain`.
+
+## Queue Trigger Deduplication
+
+The engine records each `<uuid>:<timestamp>` pair for 140 seconds after the
+UUID maps to a configured job. During that retention window, another queue
+payload with the same UUID and timestamp is acknowledged and discarded before a
+second pipeline is started.
+
+Deduplication is intentionally in the engine queue handler, not in queue
+providers, so all provider backends share the same behavior. The dedupe key is
+stored only in memory and is cleaned opportunistically as new queue messages are
+handled. The timestamp is only an idempotency token; it is not interpreted as a
+wall-clock freshness check.
 
 ## Job Matching And Pipeline Start
 
@@ -308,15 +326,18 @@ Flow:
 2. Parse and validate the queue body.
 3. Lookup the normalized UUID in the current enabled-job UUID map.
 4. If no job matches, log `Error` and return `QueueAck`.
-5. Parse shell-escaped arguments.
-6. Validate supplied arguments against the job's `Arguments` matchers:
+5. Record the `<uuid>:<timestamp>` dedupe key for 140 seconds.
+6. If the dedupe key is already present, log `Info` and return `QueueAck`
+   without starting another pipeline.
+7. Parse shell-escaped arguments.
+8. Validate supplied arguments against the job's `Arguments` matchers:
    - Too few arguments is an error.
    - Existing behavior for extra arguments is preserved.
    - Invalid argument values log `Error`.
    - Queue triggers never prompt for missing arguments.
-7. Log `Info` that job `<name>` was triggered from queue provider `<provider>`.
-8. Start the job pipeline with a new pipeline type, `queuedJob`.
-9. Return `QueueAck` once the pipeline has been accepted for execution.
+9. Log `Info` that job `<name>` was triggered from queue provider `<provider>`.
+10. Start the job pipeline with a new pipeline type, `queuedJob`.
+11. Return `QueueAck` once the pipeline has been accepted for execution.
 
 The worker should be built like a scheduled job worker:
 
@@ -390,15 +411,19 @@ Google resource alignment:
   `job-triggers` and subscription `job-triggers-pull`.
 - `resources/gcloud/scripts/create-job-webhook-resources.sh` deploys a Cloud
   Function with a UUID-bearing URL and publisher service account.
-- `resources/gcloud/scripts/webhook/index.js` publishes the raw request body.
-- `resources/gcloud/scripts/queue-job.sh` sends the body contract expected by
-  the engine.
+- `resources/gcloud/scripts/webhook/index.js` validates the raw request body
+  has the required UUID, timestamp, and argument separator shape before
+  publishing it.
+- `resources/gcloud/scripts/queue-job.sh` calculates a 14-digit timestamp and
+  sends the body contract expected by the engine.
 
 ## Security Notes
 
 - Queue provider config is engine/provider config, not extension config.
 - `UUIDTrigger` values are bearer trigger secrets and should normally be stored
   in `conf/variables/<environment>.yaml` under `Secrets`.
+- Queue timestamps are idempotency tokens paired with a UUID trigger. They are
+  not authorization material.
 - Logs must never print configured UUIDs or full queue bodies.
 - Unknown UUIDs should log only a short diagnostic such as provider name,
   provider message ID, and body length.
@@ -414,6 +439,7 @@ Google resource alignment:
 First-slice semantics are accepted-trigger acknowledgment:
 
 - Ack after the engine validates the queue body and accepts the job pipeline.
+- Ack duplicate UUID/timestamp pairs after discarding the duplicate.
 - Ack malformed bodies and unknown UUIDs after logging an error.
 - Retry only when the engine is shutting down or temporarily unable to make a
   routing decision.
@@ -431,7 +457,8 @@ Implications:
 ### 1) Change Summary
 
 - Slice name: job queue provider facility.
-- Goal: allow remote queues to trigger configured jobs by UUID and arguments.
+- Goal: allow remote queues to trigger configured jobs by UUID/timestamp
+  prefixes and arguments.
 - Out of scope: completion-coupled queue retries, user-scheduled jobs, extension
   queue APIs, and connector-based queue routing.
 
@@ -447,7 +474,8 @@ Expected implementation files and directories:
 - `bot/config_validate.go`: `conf/queues/*.yaml` validation.
 - `bot/tasks.go`: `Job.UUIDTrigger`.
 - `bot/taskconf.go`: UUID trigger parsing, validation, duplicate detection.
-- `bot/queue_runtime.go`: provider lifecycle and engine queue message handler.
+- `bot/queue_runtime.go`: provider lifecycle, queue body parsing, dedupe cache,
+  and engine queue message handler.
 - `bot/bot_process.go`: startup and shutdown placement.
 - `bot/run_pipelines.go`, `bot/constants.go`, `bot/pipelinetype_string.go`:
   queued-job pipeline source.
@@ -487,6 +515,8 @@ What changes:
 - Root config can enable queue providers.
 - Jobs can declare `UUIDTrigger`.
 - Queue providers poll external systems and submit queue messages to the engine.
+- Queue bodies carry a UUID/timestamp prefix, and the engine deduplicates
+  matching triggers with the same prefix for 140 seconds.
 - The engine starts matching jobs from queue messages with a `queuedJob`
   pipeline type.
 
@@ -532,8 +562,8 @@ What does not change:
 - Race risks: reload while queue item is being matched; provider shutdown while
   a message callback is active; duplicate queue delivery from provider restart.
 - Mitigations: immutable config snapshots, provider stop channels, retry during
-  shutdown, idempotent job design guidance, and unit tests for reload/shutdown
-  behavior.
+  shutdown, engine-side UUID/timestamp dedupe, and unit tests for
+  reload/shutdown behavior.
 
 ### 8) Backward Compatibility
 
@@ -548,10 +578,12 @@ What does not change:
 Focused tests:
 
 - Config validation accepts `conf/queues/gcloud.yaml` with `QueueConfig`.
-- Root `QueueConfig` is rejected with a migration-style error.
+- Root `QueueConfig` is rejected with a configuration-placement error.
 - `QueueProviders` parses and normalizes provider names.
 - `UUIDTrigger` accepts valid UUIDs and rejects invalid or duplicate values.
 - Queue body parsing preserves arguments with spaces.
+- Queue body parsing requires a 12-15 digit timestamp after the UUID.
+- Duplicate UUID/timestamp pairs are discarded for 140 seconds.
 - Unknown UUID logs an error and returns `QueueAck`.
 - Shutdown returns `QueueRetry`.
 - Queue-triggered jobs start with `automaticTask=true`, expected args, and
@@ -581,8 +613,7 @@ Manual verification:
 - `aidocs/SCHEDULER_FLOW.md`: no behavior change, but cross-reference queued
   jobs if helpful.
 - `aidocs/COMPONENT_MAP.md`: add `queues/`, `conf/queues/`, and this document.
-- `aidocs/V3_COMPATIBILITY_CONTRACT.md`: no required change unless config
-  migration guidance changes.
+- `aidocs/V3_COMPATIBILITY_CONTRACT.md`: no required change.
 - `conf/README.md`: add `queues/<provider>.yaml`.
 - `resources/gcloud/README.md`: document queue resource scripts and curl flow.
 

aidocs/PIPELINE_LIFECYCLE.md

@@ -134,8 +134,9 @@ Catch-all mode scoping:
 - Plugin match → `startPipeline(..., plugCommand|plugMessage, ...)`: `bot/dispatch.go:checkPluginMatchersAndRun`, `bot/constants.go` `pipelineType`.
 - Job trigger / command → `startPipeline(..., jobTrigger|jobCommand, ...)`: `bot/jobrun.go:checkJobMatchersAndRun`, `bot/constants.go` `pipelineType`.
 - Queue trigger → `triggerJobFromQueue(...)` parses the queue body, matches a
-  job `UUIDTrigger`, and starts `startPipeline(..., queuedJob, ...)` without
-  entering connector message routing: `bot/queue_runtime.go`.
+  job `UUIDTrigger`, deduplicates matching UUID/timestamp prefixes, and starts
+  `startPipeline(..., queuedJob, ...)` without entering connector message
+  routing: `bot/queue_runtime.go`.
 - `startPipeline` now stamps each pipeline with `startedAt`, effective timeout settings, and operator-channel routing metadata before the primary task runs: `bot/run_pipelines.go`, `bot/pipecontext.go`.
 - A bounded live log buffer is attached to every pipeline through `newPipelineLiveLogger(...)`: `bot/history.go`, `bot/pipeline_monitoring.go`.
   - The live buffer tees normal history logging and keeps recent section markers, `Robot.Log(...)` / `worker.Log(...)`, and child stdout/stderr even when a job/plugin later discards persisted history (`KeepLogs: 0`).

bot/queue_runtime.go

@@ -7,13 +7,24 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	shlex "github.com/anmitsu/go-shlex"
 	"github.com/google/uuid"
 	"github.com/lnxjedi/gopherbot/robot"
 )
 
 const queueUUIDPrefixLen = 36
+const queueTimestampMinLen = 12
+const queueTimestampMaxLen = 15
+
+const queueDedupeRetention = 140 * time.Second
+
+type parsedQueueBody struct {
+	jobUUID   string
+	timestamp string
+	args      []string
+}
 
 type queueHandler struct {
 	handler
@@ -38,6 +49,13 @@ var runtimeQueueProviders = struct {
 	runtimes: map[string]*managedQueueProvider{},
 }
 
+var queueDedupe = struct {
+	sync.Mutex
+	seen map[string]time.Time
+}{
+	seen: map[string]time.Time{},
+}
+
 func (h queueHandler) GetQueueConfig(v interface{}) error {
 	cfg := getQueueConfigFor(h.provider)
 	if cfg == nil {
@@ -251,29 +269,68 @@ func reconcileQueueProviderRuntimes(providers []string) {
 	}
 }
 
-func parseQueueBody(body []byte) (string, []string, error) {
+func parseQueueBody(body []byte) (parsedQueueBody, error) {
 	if len(body) < queueUUIDPrefixLen {
-		return "", nil, fmt.Errorf("queue body too short: %d byte(s)", len(body))
+		return parsedQueueBody{}, fmt.Errorf("queue body too short: %d byte(s)", len(body))
 	}
 	id, err := uuid.ParseBytes(body[:queueUUIDPrefixLen])
 	if err != nil {
-		return "", nil, fmt.Errorf("invalid queue UUID prefix: %w", err)
+		return parsedQueueBody{}, fmt.Errorf("invalid queue UUID prefix: %w", err)
+	}
+	if len(body) == queueUUIDPrefixLen || body[queueUUIDPrefixLen] != ':' {
+		return parsedQueueBody{}, fmt.Errorf("queue UUID prefix is not followed by a colon")
+	}
+
+	timestampStart := queueUUIDPrefixLen + 1
+	timestampEnd := timestampStart
+	for timestampEnd < len(body) && body[timestampEnd] >= '0' && body[timestampEnd] <= '9' {
+		timestampEnd++
 	}
-	if len(body) == queueUUIDPrefixLen {
-		return id.String(), nil, nil
+	timestampLen := timestampEnd - timestampStart
+	if timestampLen < queueTimestampMinLen || timestampLen > queueTimestampMaxLen {
+		return parsedQueueBody{}, fmt.Errorf("queue timestamp must be %d-%d digits", queueTimestampMinLen, queueTimestampMaxLen)
 	}
-	if body[queueUUIDPrefixLen] != ' ' {
-		return "", nil, fmt.Errorf("queue UUID prefix is not followed by a space")
+	if timestampEnd == len(body) {
+		return parsedQueueBody{
+			jobUUID:   id.String(),
+			timestamp: string(body[timestampStart:timestampEnd]),
+		}, nil
 	}
-	argText := strings.TrimSpace(string(body[queueUUIDPrefixLen+1:]))
+	if body[timestampEnd] != ' ' {
+		return parsedQueueBody{}, fmt.Errorf("queue timestamp is not followed by a space")
+	}
+	argText := strings.TrimSpace(string(body[timestampEnd+1:]))
 	if argText == "" {
-		return id.String(), nil, nil
+		return parsedQueueBody{
+			jobUUID:   id.String(),
+			timestamp: string(body[timestampStart:timestampEnd]),
+		}, nil
 	}
 	args, err := shlex.Split(argText, true)
 	if err != nil {
-		return "", nil, fmt.Errorf("parsing shell-escaped queue arguments: %w", err)
+		return parsedQueueBody{}, fmt.Errorf("parsing shell-escaped queue arguments: %w", err)
 	}
-	return id.String(), args, nil
+	return parsedQueueBody{
+		jobUUID:   id.String(),
+		timestamp: string(body[timestampStart:timestampEnd]),
+		args:      args,
+	}, nil
+}
+
+func recordQueueDedupe(jobUUID, timestamp string, now time.Time) bool {
+	key := jobUUID + ":" + timestamp
+	queueDedupe.Lock()
+	defer queueDedupe.Unlock()
+	for seenKey, expiresAt := range queueDedupe.seen {
+		if !now.Before(expiresAt) {
+			delete(queueDedupe.seen, seenKey)
+		}
+	}
+	if expiresAt, ok := queueDedupe.seen[key]; ok && now.Before(expiresAt) {
+		return true
+	}
+	queueDedupe.seen[key] = now.Add(queueDedupeRetention)
+	return false
 }
 
 func triggerJobFromQueue(provider string, msg robot.QueueMessage) robot.QueueDisposition {
@@ -284,7 +341,7 @@ func triggerJobFromQueue(provider string, msg robot.QueueMessage) robot.QueueDis
 	}
 	state.RUnlock()
 
-	jobUUID, args, err := parseQueueBody(msg.Body)
+	parsed, err := parseQueueBody(msg.Body)
 	if err != nil {
 		Log(robot.Error, "Queue provider '%s' message '%s' rejected: %v (body length %d)", provider, msg.ID, err, len(msg.Body))
 		return robot.QueueAck
@@ -297,14 +354,22 @@ func triggerJobFromQueue(provider string, msg robot.QueueMessage) robot.QueueDis
 	if protocol == "" {
 		protocol = currentCfg.protocol
 	}
-	taskItem := tasks.uuidTriggers[jobUUID]
+	taskItem := tasks.uuidTriggers[parsed.jobUUID]
 	currentCfg.RUnlock()
 
 	if taskItem == nil {
 		Log(robot.Error, "Queue provider '%s' message '%s' had no matching job UUID (body length %d)", provider, msg.ID, len(msg.Body))
 		return robot.QueueAck
 	}
 	task, _, job := getTask(taskItem)
+	if recordQueueDedupe(parsed.jobUUID, parsed.timestamp, time.Now()) {
+		jobName := "<unknown>"
+		if task != nil {
+			jobName = task.name
+		}
+		Log(robot.Info, "Queue provider '%s' message '%s' discarded duplicate queue trigger for job '%s'", provider, msg.ID, jobName)
+		return robot.QueueAck
+	}
 	if job == nil {
 		Log(robot.Error, "Queue provider '%s' message '%s' matched non-job task '%s'", provider, msg.ID, task.name)
 		return robot.QueueAck
@@ -313,12 +378,12 @@ func triggerJobFromQueue(provider string, msg robot.QueueMessage) robot.QueueDis
 		Log(robot.Error, "Queue provider '%s' message '%s' matched disabled job '%s'", provider, msg.ID, task.name)
 		return robot.QueueAck
 	}
-	if len(args) < len(job.Arguments) {
-		Log(robot.Error, "Queue provider '%s' message '%s' supplied too few arguments for job '%s': %d required but %d given", provider, msg.ID, task.name, len(job.Arguments), len(args))
+	if len(parsed.args) < len(job.Arguments) {
+		Log(robot.Error, "Queue provider '%s' message '%s' supplied too few arguments for job '%s': %d required but %d given", provider, msg.ID, task.name, len(job.Arguments), len(parsed.args))
 		return robot.QueueAck
 	}
 	for i, jobarg := range job.Arguments {
-		if !jobarg.re.MatchString(args[i]) {
+		if !jobarg.re.MatchString(parsed.args[i]) {
 			Log(robot.Error, "Queue provider '%s' message '%s' argument %d for job '%s' did not match configured argument pattern", provider, msg.ID, i+1, task.name)
 			return robot.QueueAck
 		}
@@ -336,6 +401,6 @@ func triggerJobFromQueue(provider string, msg robot.QueueMessage) robot.QueueDis
 		queueProvider:  provider,
 		queueMessageID: msg.ID,
 	}
-	go w.startPipeline(nil, taskItem, queuedJob, "run", args...)
+	go w.startPipeline(nil, taskItem, queuedJob, "run", parsed.args...)
 	return robot.QueueAck
 }