fix(sqs): RecognisesPartitionedKey checks prefix only, not full parse

Round 5 review on PR #715 raised a nit: RecognisesPartitionedKey
delegated to parsePartitionedSQSKey, which fails when the queue
segment is malformed base64 or the partition segment is truncated.
For such corrupt-shape keys the predicate returned false, so the
router fell through to the engine and silently routed via routeKey's
!sqs|route|global collapse to the SQS catalog default group —
exactly the failure mode the round 5 fail-closed change was
introduced to prevent.

The fix is the structural-only intent the kv.PartitionResolver
contract already documents: "Implementations answer purely on
prefix / structural inspection". Make RecognisesPartitionedKey
match the intent — accept ANY key that starts with one of the
partitioned family prefixes, regardless of subsequent corruption.
ResolveGroup still returns (0, false) for malformed keys, and the
router pairs that with Recognised=true to fail closed.

Tests

- TestSQSPartitionResolver_RecognisesMalformedPartitionedKey:
  three sub-cases pin the new contract — prefix-only, prefix +
  invalid base64 queue segment, prefix + valid queue + '|' but
  truncated partition bytes. All assert Recognised=true and
  ResolveGroup ok=false, which is exactly the fail-closed pairing
  the router consumes.
- Existing TestSQSPartitionResolver_RecognisesPartitionedKey
  cases stay valid (the well-formed shapes still match).

Author: bootjp

adapter/sqs_partition_resolver.go

@@ -99,13 +99,14 @@ func (r *SQSPartitionResolver) ResolveGroup(key []byte) (uint64, bool) {
 }
 
 // RecognisesPartitionedKey reports whether key has the structural
-// shape of a partitioned-SQS key — a partitioned family prefix
-// followed by an encoded queue segment, a '|' terminator, and a
-// fixed-width uint32 partition index. Implementations of
-// kv.PartitionResolver must answer purely on shape so the router
-// can use the predicate to decide between fall-through (not
-// partitioned) and fail-closed (partitioned but unresolved); see
-// kv.PartitionResolver doc and codex P1 round 2 on PR #715.
+// shape of a partitioned-SQS key — i.e. starts with one of the
+// partitioned family prefixes. The check is PREFIX-ONLY, not a
+// full parse: a key with a partitioned prefix followed by a
+// malformed queue / partition segment still answers true, so the
+// router fails closed via kv.PartitionResolver semantics instead
+// of falling through to the engine and silently routing to the
+// SQS catalog default group via routeKey's !sqs|route|global
+// collapse (round 5 review nit on PR #715).
 //
 // A nil receiver returns false so kv.ShardRouter's typed-nil case
 // (ResolveGroup(nil) == (0, false)) pairs with an honest "I don't
@@ -114,7 +115,7 @@ func (r *SQSPartitionResolver) RecognisesPartitionedKey(key []byte) bool {
 	if r == nil || len(key) == 0 {
 		return false
 	}
-	_, _, ok := parsePartitionedSQSKey(key)
+	_, ok := stripPartitionedFamilyPrefix(key)
 	return ok
 }
 

adapter/sqs_partition_resolver_test.go

@@ -280,6 +280,57 @@ func TestSQSPartitionResolver_RecognisesPartitionedKey_NilReceiver(t *testing.T)
 	require.False(t, r.RecognisesPartitionedKey([]byte("any-key")))
 }
 
+// TestSQSPartitionResolver_RecognisesMalformedPartitionedKey pins
+// the prefix-only check (round 5 nit on PR #715): a key with a
+// valid partitioned family prefix but a corrupt / truncated queue
+// or partition segment MUST still be recognised so the router
+// fails closed rather than falling through to engine routing
+// (which would silently mis-route to the SQS catalog default
+// group via routeKey's !sqs|route|global collapse).
+func TestSQSPartitionResolver_RecognisesMalformedPartitionedKey(t *testing.T) {
+	t.Parallel()
+	r := NewSQSPartitionResolver(map[string][]uint64{
+		"q.fifo": {10, 11},
+	})
+	cases := []struct {
+		name string
+		key  []byte
+	}{
+		{
+			// Prefix only, nothing after — parsePartitionedSQSKey
+			// would fail on the missing '|' terminator, but the
+			// shape IS recognised.
+			name: "prefix only",
+			key:  []byte(SqsPartitionedMsgDataPrefix),
+		},
+		{
+			// Prefix + base64 garbage that decodes invalidly.
+			// parsePartitionedSQSKey would fail at decodeSQSSegment.
+			name: "prefix + invalid base64",
+			key:  []byte(SqsPartitionedMsgDataPrefix + "!!!|"),
+		},
+		{
+			// Prefix + valid queue segment + '|' but no partition
+			// bytes (truncated).
+			name: "prefix + queue + truncated partition",
+			key:  []byte(SqsPartitionedMsgVisPrefix + encodeSQSSegment("q.fifo") + "|"),
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			require.True(t, r.RecognisesPartitionedKey(tc.key),
+				"malformed partitioned key MUST be recognised so "+
+					"the router fails closed rather than mis-routing "+
+					"through the engine's SQS catalog default")
+			gid, ok := r.ResolveGroup(tc.key)
+			require.False(t, ok,
+				"malformed key cannot resolve")
+			require.Equal(t, uint64(0), gid)
+		})
+	}
+}
+
 // TestSQSPartitionResolver_PrefixesAlign pins that the resolver's
 // family-prefix list matches the constants in sqs_keys.go exactly.
 // A future renamed prefix or added family that touches sqs_keys.go